Fixes proxy startup

proxy/addon.py

@@ -7,14 +7,17 @@
 import json
 import os
 import re
-from dataclasses import dataclass
 from pathlib import Path
 
+from container import ContainerMetadata, ContainerResolver
 from db import ProxyDB, get_db_path
 from mitmproxy import ctx, http, websocket
 from policy import FilterDecision, PolicyStore
 
-_DEFAULT_MAPPING_PATH = Path("/data/containers.json")
+# NOTE: mitmproxy runs this file as a script module that is absent from
+# sys.modules, which breaks @dataclass on classes with string annotations.
+# Define dataclasses in a sibling module (see container.py) and import them.
+
 _POLICY_USERNAME_RE = re.compile(r"^vp-([0-9a-f]{32})$")
 
 
@@ -48,53 +51,6 @@ def _pop_policy_identity(flow: http.HTTPFlow) -> tuple[str | None, bool]:
     return (match.group(1), False) if match else (None, True)
 
 
-@dataclass(frozen=True)
-class ContainerMetadata:
-    container_id: str | None = None
-    container_name: str | None = None
-    policy_id: str | None = None
-    profile: str | None = None
-
-
-class ContainerResolver:
-    """Resolves client IPs to container metadata via a shared JSON file."""
-
-    def __init__(self, path: Path = _DEFAULT_MAPPING_PATH) -> None:
-        self._path = path
-        self._mtime: float = 0.0
-        self._mapping: dict[str, dict[str, str]] = {}
-
-    def resolve(self, ip: str | None) -> ContainerMetadata:
-        """Return mapped metadata for the given client IP."""
-        if ip is None:
-            return ContainerMetadata()
-        self._maybe_reload()
-        entry = self._mapping.get(ip)
-        if entry is None:
-            return ContainerMetadata()
-        return ContainerMetadata(
-            container_id=entry.get("container_id"),
-            container_name=entry.get("container_name"),
-            policy_id=entry.get("policy_id"),
-            profile=entry.get("profile"),
-        )
-
-    def _maybe_reload(self) -> None:
-        try:
-            st = os.stat(self._path)
-        except OSError:
-            return
-        if st.st_mtime == self._mtime:
-            return
-        try:
-            data = json.loads(self._path.read_text())
-            if isinstance(data, dict):
-                self._mapping = data
-            self._mtime = st.st_mtime
-        except (json.JSONDecodeError, OSError):
-            pass
-
-
 class SQLiteLogger:
     def __init__(self) -> None:
         self._db: ProxyDB | None = None

proxy/container.py

@@ -0,0 +1,64 @@
+"""Client-IP to container metadata resolution.
+
+Lives outside addon.py on purpose: mitmproxy executes the addon script with a
+module object it never puts in `sys.modules`, and `@dataclass` on a class with
+string annotations (`from __future__ import annotations`) looks its own module
+up there while processing fields. Regular imports like this one get a real
+`sys.modules` entry, so dataclasses defined here work.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+_DEFAULT_MAPPING_PATH = Path("/data/containers.json")
+
+
+@dataclass(frozen=True)
+class ContainerMetadata:
+    container_id: str | None = None
+    container_name: str | None = None
+    policy_id: str | None = None
+    profile: str | None = None
+
+
+class ContainerResolver:
+    """Resolves client IPs to container metadata via a shared JSON file."""
+
+    def __init__(self, path: Path = _DEFAULT_MAPPING_PATH) -> None:
+        self._path = path
+        self._mtime: float = 0.0
+        self._mapping: dict[str, dict[str, str]] = {}
+
+    def resolve(self, ip: str | None) -> ContainerMetadata:
+        """Return mapped metadata for the given client IP."""
+        if ip is None:
+            return ContainerMetadata()
+        self._maybe_reload()
+        entry = self._mapping.get(ip)
+        if entry is None:
+            return ContainerMetadata()
+        return ContainerMetadata(
+            container_id=entry.get("container_id"),
+            container_name=entry.get("container_name"),
+            policy_id=entry.get("policy_id"),
+            profile=entry.get("profile"),
+        )
+
+    def _maybe_reload(self) -> None:
+        try:
+            st = os.stat(self._path)
+        except OSError:
+            return
+        if st.st_mtime == self._mtime:
+            return
+        try:
+            data = json.loads(self._path.read_text())
+            if isinstance(data, dict):
+                self._mapping = data
+            self._mtime = st.st_mtime
+        except (json.JSONDecodeError, OSError):
+            pass

tests/test_addon.py

@@ -9,6 +9,7 @@
 from pathlib import Path
 
 import addon as addon_mod
+import container as container_mod
 from mitmproxy import http
 from mitmproxy.test import taddons, tflow
 
@@ -317,7 +318,7 @@ def test_source_mapping_overrides_conflicting_supplied_identity(
             },
         ),
     )
-    monkeypatch.setattr(addon_mod, "_DEFAULT_MAPPING_PATH", tmp_path / "containers.json")
+    monkeypatch.setattr(container_mod, "_DEFAULT_MAPPING_PATH", tmp_path / "containers.json")
     with taddons.context(logger):
         logger.load(None)
         assert logger._resolver is not None

tests/test_addon_load.py

@@ -0,0 +1,50 @@
+"""Load addon.py the way mitmproxy does, not the way pytest does.
+
+mitmproxy's script loader (`mitmproxy.addons.script.load_script`) builds the
+addon module with `importlib.util.module_from_spec` + `exec_module` and never
+registers it in `sys.modules`. Anything executed at import time that looks the
+module up by `__module__` therefore sees `None` -- most notably
+`@dataclass` on a class with string annotations, which crashes in
+`dataclasses._is_type`. A plain `import addon` hides that entirely.
+"""
+
+from __future__ import annotations
+
+import importlib.machinery
+import importlib.util
+import os
+import sys
+import types
+from pathlib import Path
+
+ADDON_PATH = Path(__file__).resolve().parents[1] / "proxy" / "addon.py"
+
+
+def _load_like_mitmproxy(path: Path) -> types.ModuleType:
+    """Mirror mitmproxy.addons.script.load_script, minus the error swallowing."""
+    fullname = f"__mitmproxy_script__.{path.stem}"
+    sys.modules.pop(fullname, None)
+    oldpath = sys.path
+    sys.path = [str(path.parent), *sys.path]
+    try:
+        loader = importlib.machinery.SourceFileLoader(fullname, str(path))
+        spec = importlib.util.spec_from_loader(fullname, loader=loader)
+        assert spec is not None
+        module = importlib.util.module_from_spec(spec)
+        loader.exec_module(module)
+        return module
+    finally:
+        sys.path[:] = oldpath
+        sys.modules.pop(fullname, None)
+
+
+def test_addon_imports_outside_sys_modules() -> None:
+    module = _load_like_mitmproxy(ADDON_PATH)
+    assert hasattr(module, "SQLiteLogger")
+    assert module.__name__ not in sys.modules
+
+
+def test_addon_module_is_not_registered_by_loader() -> None:
+    """Guard the assumption the other test rests on."""
+    assert "__mitmproxy_script__.addon" not in sys.modules
+    assert os.path.exists(ADDON_PATH)