[Codex] fix(logic): harden formula eval against sandbox escape

frontend/src/composables/useIcons.test.ts

@@ -0,0 +1,45 @@
+// @vitest-environment jsdom
+import { describe, expect, it, vi } from 'vitest'
+
+async function loadUseIconsWithSvg(svgContent: string) {
+  vi.resetModules()
+  vi.doMock('@/api/client', () => ({
+    icons: {
+      list: vi.fn().mockResolvedValue({
+        icons: [{ name: 'test-icon', content: svgContent }],
+      }),
+    },
+  }))
+  const mod = await import('./useIcons')
+  return mod.useIcons()
+}
+
+describe('useIcons sanitizeSvg', () => {
+  it('blocks javascript href with embedded control chars', async () => {
+    const { getSvg } = await loadUseIconsWithSvg(
+      '<svg xmlns="http://www.w3.org/2000/svg"><a href="java&#x0A;script:alert(1)"><rect width="12" height="8"/></a></svg>',
+    )
+
+    const sanitized = await getSvg('test-icon')
+    expect(sanitized).not.toContain('href=')
+  })
+
+  it('still removes root width/height attributes', async () => {
+    const { getSvg } = await loadUseIconsWithSvg(
+      '<svg xmlns="http://www.w3.org/2000/svg" width="100" height="50"><rect width="12" height="8"/></svg>',
+    )
+
+    const sanitized = await getSvg('test-icon')
+    expect(sanitized).not.toContain(' width="100"')
+    expect(sanitized).not.toContain(' height="50"')
+  })
+
+  it('removes SMIL animation elements that can mutate links', async () => {
+    const { getSvg } = await loadUseIconsWithSvg(
+      '<svg xmlns="http://www.w3.org/2000/svg"><a href="#"><animate attributeName="href" values="javascript:alert(1)"/></a></svg>',
+    )
+
+    const sanitized = await getSvg('test-icon')
+    expect(sanitized).not.toContain('<animate')
+  })
+})

frontend/src/composables/useIcons.ts

@@ -6,6 +6,15 @@ const iconNames = ref<string[]>([])
 const svgCache: Record<string, string> = {}  // name → normalised SVG string
 let listPromise: Promise<void> | null = null
 
+function isDangerousHref(raw: string): boolean {
+  // Normalize obfuscated schemes like "java\nscript:" before checking.
+  const normalized = raw
+    .trim()
+    .toLowerCase()
+    .replace(/[\u0000-\u001f\u007f\s]+/g, '')
+  return normalized.startsWith('javascript:')
+}
+
 function sanitizeSvg(raw: string): string {
   const parser = new DOMParser()
   const doc = parser.parseFromString(raw, 'image/svg+xml')
@@ -14,12 +23,14 @@ function sanitizeSvg(raw: string): string {
 
   // Remove executable or HTML-capable elements
   doc.querySelectorAll('script, foreignObject').forEach((el) => el.remove())
+  // Remove SMIL animation elements that can mutate attributes post-sanitization.
+  doc.querySelectorAll('animate, animateMotion, animateTransform, set').forEach((el) => el.remove())
 
   // Remove dangerous attributes and fixed dimensions
   for (const el of Array.from(doc.querySelectorAll('*'))) {
     for (const attr of Array.from(el.attributes)) {
       const name = attr.name.toLowerCase()
-      const value = attr.value.trim().toLowerCase()
+      const value = attr.value
 
       if (name === 'width' || name === 'height') {
         el.removeAttribute(attr.name)
@@ -31,7 +42,7 @@ function sanitizeSvg(raw: string): string {
         continue
       }
 
-      if ((name === 'href' || name === 'xlink:href') && value.startsWith('javascript:')) {
+      if ((name === 'href' || name === 'xlink:href') && isDangerousHref(value)) {
         el.removeAttribute(attr.name)
       }
     }

obs/config.py

@@ -121,6 +121,11 @@ def _has_env_key_case_insensitive(env_key: str) -> bool:
     return any(existing.upper() == lookup for existing in os.environ)
 
 
+def _get_existing_env_key_case_insensitive(env_key: str) -> str | None:
+    lookup = env_key.upper()
+    return next((existing for existing in os.environ if existing.upper() == lookup), None)
+
+
 def _get_env_case_insensitive(*env_keys: str) -> str | None:
     for key in env_keys:
         if key in os.environ:
@@ -133,16 +138,30 @@ def _get_env_case_insensitive(*env_keys: str) -> str | None:
 
 
 def _import_legacy_env_vars() -> None:
-    """Import OPENTWS_* variables as OBS_* when OBS_* is not set."""
+    """Import OPENTWS_* variables as OBS_* when OBS_* is not set.
+
+    Docker images may predefine OBS_* defaults. If those exact defaults are
+    present, prefer an explicitly supplied OPENTWS_* value for compatibility.
+    """
     legacy_prefix = "OPENTWS_"
     new_prefix = "OBS_"
+    docker_defaults = {
+        "OBS_CONFIG": "/data/config.yaml",
+        "OBS_DATABASE__PATH": "/data/obs.db",
+    }
 
     for key, value in list(os.environ.items()):
         if not key.startswith(legacy_prefix):
             continue
         mapped_key = f"{new_prefix}{key[len(legacy_prefix) :]}"
-        if not _has_env_key_case_insensitive(mapped_key):
+        existing_key = _get_existing_env_key_case_insensitive(mapped_key)
+        if existing_key is None:
             os.environ[mapped_key] = value
+            continue
+
+        expected_default = docker_defaults.get(mapped_key.upper())
+        if expected_default is not None and os.environ.get(existing_key) == expected_default:
+            os.environ[existing_key] = value
 
 
 def _resolve_default_db_path(default_path: str = "/data/obs.db") -> str:

obs/logic/executor.py

@@ -1123,10 +1123,68 @@ def _safe_eval(expr: str, ctx: dict[str, Any]) -> Any:
         allowed.update(ctx)
         try:
             tree = ast.parse(expr, mode="eval")
+            GraphExecutor._validate_formula_ast(tree, set(allowed.keys()))
             return eval(compile(tree, "<formula>", "eval"), {"__builtins__": {}}, allowed)  # noqa: S307
         except Exception as exc:
             raise ExecutionError(f"Formula error: {exc}") from exc
 
+    @staticmethod
+    def _validate_formula_ast(tree: ast.AST, allowed_names: set[str]) -> None:
+        """Validate formula AST and reject unsafe Python constructs."""
+        allowed_nodes = (
+            ast.Expression,
+            ast.BinOp,
+            ast.UnaryOp,
+            ast.BoolOp,
+            ast.Compare,
+            ast.IfExp,
+            ast.Call,
+            ast.keyword,
+            ast.Name,
+            ast.Load,
+            ast.Constant,
+            ast.List,
+            ast.Tuple,
+            ast.Dict,
+            ast.Set,
+            ast.Subscript,
+            ast.Slice,
+            ast.Index,
+            ast.Add,
+            ast.Sub,
+            ast.Mult,
+            ast.Div,
+            ast.FloorDiv,
+            ast.Mod,
+            ast.Pow,
+            ast.UAdd,
+            ast.USub,
+            ast.And,
+            ast.Or,
+            ast.Not,
+            ast.Eq,
+            ast.NotEq,
+            ast.Lt,
+            ast.LtE,
+            ast.Gt,
+            ast.GtE,
+            ast.Is,
+            ast.IsNot,
+            ast.In,
+            ast.NotIn,
+        )
+
+        for node in ast.walk(tree):
+            if not isinstance(node, allowed_nodes):
+                raise ExecutionError(f"Disallowed expression element: {type(node).__name__}")
+            if isinstance(node, ast.Name) and node.id not in allowed_names:
+                raise ExecutionError(f"Unknown variable or function: {node.id}")
+            if isinstance(node, ast.Call):
+                if not isinstance(node.func, ast.Name):
+                    raise ExecutionError("Only direct function calls are allowed")
+                if node.func.id not in allowed_names:
+                    raise ExecutionError(f"Function not allowed: {node.func.id}")
+
     @staticmethod
     def _run_script(script: str, inputs: dict[str, Any]) -> Any:
         """Run a restricted Python script."""

tests/unit/test_config_legacy_env.py

@@ -0,0 +1,39 @@
+import os
+
+from obs.config import _import_legacy_env_vars
+
+
+def test_legacy_config_overrides_docker_default_obs_config(monkeypatch):
+    monkeypatch.setenv("OBS_CONFIG", "/data/config.yaml")
+    monkeypatch.setenv("OPENTWS_CONFIG", "/legacy/config.yaml")
+
+    _import_legacy_env_vars()
+
+    assert os.environ["OBS_CONFIG"] == "/legacy/config.yaml"
+
+
+def test_legacy_db_path_overrides_docker_default_obs_db_path(monkeypatch):
+    monkeypatch.setenv("OBS_DATABASE__PATH", "/data/obs.db")
+    monkeypatch.setenv("OPENTWS_DATABASE__PATH", "/legacy/opentws.db")
+
+    _import_legacy_env_vars()
+
+    assert os.environ["OBS_DATABASE__PATH"] == "/legacy/opentws.db"
+
+
+def test_explicit_obs_config_is_not_overwritten(monkeypatch):
+    monkeypatch.setenv("OBS_CONFIG", "/custom/obs.yaml")
+    monkeypatch.setenv("OPENTWS_CONFIG", "/legacy/config.yaml")
+
+    _import_legacy_env_vars()
+
+    assert os.environ["OBS_CONFIG"] == "/custom/obs.yaml"
+
+
+def test_legacy_import_when_obs_key_missing(monkeypatch):
+    monkeypatch.delenv("OBS_CONFIG", raising=False)
+    monkeypatch.setenv("OPENTWS_CONFIG", "/legacy/config.yaml")
+
+    _import_legacy_env_vars()
+
+    assert os.environ["OBS_CONFIG"] == "/legacy/config.yaml"

tests/unit/test_executor.py

@@ -115,11 +115,30 @@ def test_builtins_blocked(self):
         with pytest.raises(ExecutionError):
             GraphExecutor._safe_eval("open('secret')", {})
 
-    def test_attribute_access_allowed(self):
-        # NOTE: current sandbox does not block __class__ access
-        # This documents the actual behaviour (not a security guarantee)
-        result = GraphExecutor._safe_eval("().__class__.__bases__", {})
-        assert result is not None  # returns (<class 'object'>,)
+    def test_attribute_access_blocked(self):
+        with pytest.raises(ExecutionError):
+            GraphExecutor._safe_eval("().__class__.__bases__", {})
+
+    def test_subclasses_escape_blocked(self):
+        payload = "[c for c in ().__class__.__base__.__subclasses__() if c.__name__=='BuiltinImporter'][0]"
+        with pytest.raises(ExecutionError):
+            GraphExecutor._safe_eval(payload, {})
+
+    def test_indirect_function_call_blocked(self):
+        with pytest.raises(ExecutionError):
+            GraphExecutor._safe_eval("([abs][0])(1)", {})
+
+    def test_keyword_arguments_allowed_for_safe_calls(self):
+        result = GraphExecutor._safe_eval("round(x, ndigits=1)", {"x": 21.15})
+        assert result == pytest.approx(21.2)
+
+    def test_is_none_comparison_allowed(self):
+        assert GraphExecutor._safe_eval("0 if x is None else x", {"x": None}) == 0
+        assert GraphExecutor._safe_eval("0 if x is None else x", {"x": 7}) == 7
+
+    def test_in_membership_comparison_allowed(self):
+        assert GraphExecutor._safe_eval("1 if x in [1, 2] else 0", {"x": 2}) == 1
+        assert GraphExecutor._safe_eval("1 if x in [1, 2] else 0", {"x": 5}) == 0
 
 
 # ===========================================================================