feat(governance): standing-grant revocation dashboard (#620)

coworker/config.py

@@ -155,3 +155,49 @@ def load_config(
                     )
                 )
     return cfg
+
+
+def revoke_allowed_domain(domain: str) -> bool:
+    """Remove one global domain while preserving every other TOML setting."""
+    import copy
+    import json
+    import os
+    import re
+    import tempfile
+
+    path = global_config_path()
+    if not path.exists():
+        return False
+    content = path.read_text(encoding="utf-8")
+    parsed = tomllib.loads(content)
+    domains = parsed.get("allowed_domains", [])
+    if not isinstance(domains, list) or domain not in domains:
+        return False
+    expected = copy.deepcopy(parsed)
+    expected["allowed_domains"] = [d for d in domains if d != domain]
+    replacement = "allowed_domains = " + json.dumps(expected["allowed_domains"], ensure_ascii=False)
+    # Consider complete array spans, then verify the entire parsed document. This
+    # rejects matches in comments, strings or nested tables without rewriting them.
+    pattern = re.compile(r"(?m)^[ \t]*(?:allowed_domains|\"allowed_domains\"|'allowed_domains')[ \t]*=")
+    for match in pattern.finditer(content):
+        for end in range(match.end(), len(content)):
+            if content[end] != "]":
+                continue
+            candidate = content[:match.start()] + replacement + content[end + 1:]
+            try:
+                if tomllib.loads(candidate) != expected:
+                    continue
+            except tomllib.TOMLDecodeError:
+                continue
+            with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, delete=False) as f:
+                tmp = Path(f.name)
+                f.write(candidate)
+            try:
+                os.chmod(tmp, path.stat().st_mode & 0o777)
+                if path.read_text(encoding="utf-8") != content:
+                    raise ValueError("config changed during revocation; retry")
+                tmp.replace(path)
+            finally:
+                tmp.unlink(missing_ok=True)
+            return True
+    raise ValueError("could not safely update allowed_domains in config.toml")

coworker/conversations.py

@@ -485,6 +485,7 @@ def list(self, *, workspace: Optional[str] = None) -> list[SessionRecord]:
                 archived=bool(r["archived"]),
                 origin=r["origin"],
                 origin_label=r["origin_label"],
+                grants=_load_grants(r["grants"] if "grants" in r.keys() else None),
                 team=_load_grants(r["team"] if "team" in r.keys() else None),
             )
             for r in rows

coworker/overrides.py

@@ -150,11 +150,13 @@ def set_trust(self, pattern: str) -> None:
             self._trust.append(pattern)
             self.save()
 
-    def revoke_trust(self, pattern: str) -> None:
+    def revoke_trust(self, pattern: str) -> bool:
         before = len(self._trust)
         self._trust = [p for p in self._trust if p != pattern]
         if len(self._trust) != before:
             self.save()
+            return True
+        return False
 
     def trust_patterns(self) -> list[str]:
         return list(self._trust)

coworker/permissions.py

@@ -559,6 +559,33 @@ def allow_domain_for_session(self, url_or_domain: str) -> None:
         if host:
             self.session_allow_domains.add(host)
 
+    def revoke_tool_for_session(self, tool_name: str) -> bool:
+        if tool_name in self.session_allow_tools:
+            self.session_allow_tools.remove(tool_name)
+            return True
+        return False
+
+    def revoke_command_for_session(self, command: str) -> bool:
+        if command in self.session_allow_commands:
+            self.session_allow_commands.remove(command)
+            return True
+        return False
+
+    def revoke_readonly_for_session(self) -> bool:
+        if self.session_readonly:
+            self.session_readonly = False
+            return True
+        return False
+
+    def revoke_domain_for_session(self, url_or_domain: str) -> bool:
+        host = _host_of(url_or_domain)
+        if host.startswith("www."):
+            host = host[4:]
+        if host in self.session_allow_domains:
+            self.session_allow_domains.remove(host)
+            return True
+        return False
+
     # -- helpers ----------------------------------------------------------------
     def _candidate(self, path: str) -> Path:
         # Relative paths resolve against the primary (workspace_root); absolute/`~` taken as-is.

coworker/server/app.py

@@ -694,6 +694,19 @@ def set_workspace_trust(body: dict) -> dict[str, Any]:
             trusted=bool((body or {}).get("trusted", False)),
         )
 
+    @app.get("/v1/grants")
+    def active_grants() -> dict[str, Any]:
+        return {"grants": manager.list_active_grants()}
+
+    @app.post("/v1/grants/revoke")
+    def revoke_grant(body: dict) -> dict[str, Any]:
+        return manager.revoke_grant(
+            grant_id=(body or {}).get("grant_id"),
+            kind=(body or {}).get("kind"),
+            target=(body or {}).get("target"),
+            source_id=(body or {}).get("source_id"),
+        )
+
     @app.post("/v1/workspaces/temp")
     def provision_temp_workspace(body: dict) -> dict[str, Any]:
         # UX-029: a code-family session starting "in a temporary folder" — created only