feat: redesign privacy API - mode enum off/audit/redact, fix stats and audit response format

- Replace boolean privacy_enabled with three-mode enum: off/audit/redact
- GET /api/v1/privacy/mode now returns {mode} instead of {enabled}
- POST/PUT /api/v1/privacy/mode validates against allowed set, returns 400 on invalid
- GET /api/v1/privacy/stats now returns rich PrivacyStats matching frontend TS types:
  requests, files, pii_by_type, actions{allow,block,redact}, top_domains[]
- GET /api/v1/privacy/audit now returns {logs, total, page} instead of {entries, total}
  pii_types is parsed from JSON string to dict in response
- Add get_mode() helper for addon.py internal consumption
- Add tests/test_privacy_api_redesign.py with 5 tests covering all changes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

services/privacy_gateway/privacy_api.py

@@ -0,0 +1,178 @@
+"""Privacy Gateway REST API — aiohttp server on port 8889.
+
+Provides /api/v1/privacy/* endpoints consumed by the KVM WebUI frontend.
+Mode enum: "off" | "audit" | "redact"
+"""
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+
+from aiohttp import web
+
+from .audit_logger import AuditLogger
+from .cert_manager import get_ca_pem
+
+log = logging.getLogger("privacy-api")
+
+_STATE_PATH = Path("/var/lib/kvm-privacy/state.json")
+_VALID_MODES = {"off", "audit", "redact"}
+
+
+def _load_state() -> dict:
+    try:
+        return json.loads(_STATE_PATH.read_text())
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {"mode": "off"}
+
+
+def _save_state(state: dict) -> None:
+    _STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
+    _STATE_PATH.write_text(json.dumps(state))
+
+
+class PrivacyAPIServer:
+    def __init__(self, host: str = "0.0.0.0", port: int = 8889) -> None:
+        self._host = host
+        self._port = port
+        self._audit = AuditLogger()
+        self._state = _load_state()
+        self._runner: web.AppRunner | None = None
+
+    async def start(self) -> None:
+        app = web.Application(middlewares=[self._cors_middleware])
+        app.router.add_get("/api/v1/privacy/mode",    self._get_mode)
+        app.router.add_post("/api/v1/privacy/mode",   self._set_mode)
+        app.router.add_put("/api/v1/privacy/mode",    self._set_mode)
+        app.router.add_get("/api/v1/privacy/stats",   self._get_stats)
+        app.router.add_get("/api/v1/privacy/audit",   self._get_audit)
+        app.router.add_get("/api/v1/privacy/cert",    self._get_cert)
+
+        self._runner = web.AppRunner(app)
+        await self._runner.setup()
+        site = web.TCPSite(self._runner, self._host, self._port)
+        await site.start()
+        log.info("Privacy API listening on %s:%s", self._host, self._port)
+
+    async def stop(self) -> None:
+        if self._runner:
+            await self._runner.cleanup()
+
+    def get_mode(self) -> str:
+        """Return current mode string — called by addon.py to check state."""
+        return self._state.get("mode", "off")
+
+    # ── middleware ──────────────────────────────────────────
+
+    @web.middleware
+    async def _cors_middleware(self, request: web.Request, handler):
+        if request.method == "OPTIONS":
+            resp = web.Response(status=204)
+        else:
+            try:
+                resp = await handler(request)
+            except web.HTTPException as exc:
+                resp = exc
+        resp.headers["Access-Control-Allow-Origin"] = "*"
+        resp.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, OPTIONS"
+        resp.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
+        return resp
+
+    # ── handlers ───────────────────────────────────────────
+
+    async def _get_mode(self, _request: web.Request) -> web.Response:
+        return web.json_response({"mode": self._state.get("mode", "off")})
+
+    async def _set_mode(self, request: web.Request) -> web.Response:
+        body = await request.json()
+        mode = body.get("mode", "")
+        if mode not in _VALID_MODES:
+            return web.json_response(
+                {"error": f"mode must be one of {sorted(_VALID_MODES)}"},
+                status=400,
+            )
+        self._state["mode"] = mode
+        _save_state(self._state)
+        log.info("Privacy mode set to %s", mode)
+        return web.json_response({"mode": mode})
+
+    async def _get_stats(self, _request: web.Request) -> web.Response:
+        raw = self._audit.stats_today()
+        by_action = raw.get("by_action", {})
+        by_domain = raw.get("by_domain", {})
+        return web.json_response({
+            "requests":    raw.get("requests", 0),
+            "files":       raw.get("files", 0),
+            "pii_by_type": raw.get("pii_by_type", {}),
+            "actions": {
+                "allow":  by_action.get("allow", 0),
+                "block":  by_action.get("block", 0),
+                "redact": by_action.get("redact", 0),
+            },
+            "top_domains": [
+                {"domain": d, "count": c}
+                for d, c in sorted(by_domain.items(), key=lambda x: -x[1])
+            ],
+        })
+
+    async def _get_audit(self, request: web.Request) -> web.Response:
+        page = int(request.query.get("page", "1"))
+        page_size = int(request.query.get("page_size", "20"))
+        domain = request.query.get("domain") or None
+        offset = (page - 1) * page_size
+
+        rows = self._audit.query(limit=page_size, offset=offset, domain=domain)
+        total = self._audit.count(domain=domain)
+
+        logs = []
+        for r in rows:
+            pii_types: dict = {}
+            if r.get("pii_types"):
+                try:
+                    pii_types = json.loads(r["pii_types"])
+                except (json.JSONDecodeError, TypeError):
+                    pass
+            logs.append({
+                "id":          r["id"],
+                "ts":          r.get("ts", ""),
+                "domain":      r.get("domain", ""),
+                "action":      r.get("action", ""),
+                "pii_types":   pii_types,
+                "filename":    r.get("filename", ""),
+                "file_size":   r.get("file_size", 0),
+                "client_ip":   r.get("client_ip", ""),
+                "request_url": r.get("request_url", ""),
+            })
+
+        return web.json_response({"logs": logs, "total": total, "page": page})
+
+    async def _get_cert(self, _request: web.Request) -> web.Response:
+        try:
+            pem = get_ca_pem()
+        except Exception as exc:
+            log.error("Failed to read CA cert: %s", exc)
+            return web.Response(status=500, text="CA certificate not available")
+        return web.Response(
+            body=pem,
+            content_type="application/x-pem-file",
+            headers={"Content-Disposition": 'attachment; filename="kvm-privacy-ca.crt"'},
+        )
+
+
+async def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(name)s %(message)s")
+    server = PrivacyAPIServer()
+    await server.start()
+    import asyncio
+    try:
+        await asyncio.Event().wait()
+    except (KeyboardInterrupt, asyncio.CancelledError):
+        pass
+    finally:
+        await server.stop()
+
+
+if __name__ == "__main__":
+    import asyncio
+    asyncio.run(main())

services/privacy_gateway/tests/test_privacy_api_redesign.py

@@ -0,0 +1,86 @@
+"""Tests for redesigned Privacy API with mode enum."""
+import json
+import pytest
+from unittest.mock import MagicMock, AsyncMock, patch
+
+
+def make_server():
+    """Create a PrivacyAPIServer instance with mocked audit logger."""
+    # Direct instantiation bypassing __init__ filesystem calls
+    from services.privacy_gateway.privacy_api import PrivacyAPIServer
+    server = object.__new__(PrivacyAPIServer)
+    server._state = {"mode": "off"}
+    server._audit = MagicMock()
+    server._runner = None
+    return server
+
+
+@pytest.mark.asyncio
+async def test_get_mode_returns_enum():
+    server = make_server()
+    req = MagicMock()
+    resp = await server._get_mode(req)
+    data = json.loads(resp.body)
+    assert "mode" in data
+    assert data["mode"] in ("off", "audit", "redact")
+    assert "enabled" not in data  # old bool field must be gone
+
+
+@pytest.mark.asyncio
+async def test_set_mode_accepts_redact():
+    server = make_server()
+    req = MagicMock()
+    req.json = AsyncMock(return_value={"mode": "redact"})
+    with patch("services.privacy_gateway.privacy_api._save_state"):
+        resp = await server._set_mode(req)
+    data = json.loads(resp.body)
+    assert data["mode"] == "redact"
+    assert server._state["mode"] == "redact"
+
+
+@pytest.mark.asyncio
+async def test_set_mode_rejects_invalid():
+    server = make_server()
+    req = MagicMock()
+    req.json = AsyncMock(return_value={"mode": "invalid_value"})
+    resp = await server._set_mode(req)
+    assert resp.status == 400
+
+
+@pytest.mark.asyncio
+async def test_get_stats_returns_rich_format():
+    server = make_server()
+    server._audit.stats_today.return_value = {
+        "requests": 10, "files": 5, "pii_total": 3,
+        "by_action": {"auto_redact": 2, "bypass": 8},
+        "by_domain": {"api.openai.com": 5, "claude.ai": 3},
+        "pii_by_type": {"id_card": 2, "phone": 1},
+    }
+    req = MagicMock()
+    resp = await server._get_stats(req)
+    data = json.loads(resp.body)
+    assert data["requests"] == 10
+    assert data["pii_by_type"]["id_card"] == 2
+    assert "actions" in data
+    assert "top_domains" in data
+    assert data["top_domains"][0]["domain"] == "api.openai.com"
+
+
+@pytest.mark.asyncio
+async def test_get_audit_returns_logs_field():
+    server = make_server()
+    server._audit.query.return_value = [{
+        "id": 1, "ts": "2026-03-03T10:00:00", "domain": "openai.com",
+        "pii_types": '{"id_card":1}', "action": "redact",
+        "filename": "doc.pdf", "file_size": 1024,
+        "client_ip": "127.0.0.1", "request_url": "/v1/files",
+    }]
+    server._audit.count.return_value = 1
+    req = MagicMock()
+    req.query = {}
+    resp = await server._get_audit(req)
+    data = json.loads(resp.body)
+    assert "logs" in data           # must be "logs" not "entries"
+    assert "entries" not in data    # old field must be gone
+    assert data["total"] == 1
+    assert isinstance(data["logs"][0]["pii_types"], dict)  # parsed, not string