From c624db5dd7f7a299e2f92239435b1e1a6035614d Mon Sep 17 00:00:00 2001 From: igor04091968 Date: Tue, 12 May 2026 01:11:01 +0300 Subject: [PATCH] feat(dlp-policy): add approval workflow, full audit trail, and CRUD documentation --- aw-server/dlp-policy-engine/policy_schema.py | 6 +- aw-server/dlp-policy-engine/policy_service.py | 43 +++++- aw-server/dlp-policy-engine/policy_storage.py | 111 ++++++++++++++-- docs/dlp-policy-engine.md | 122 ++++++++---------- 4 files changed, 203 insertions(+), 79 deletions(-) diff --git a/aw-server/dlp-policy-engine/policy_schema.py b/aw-server/dlp-policy-engine/policy_schema.py index 56c80d4..3109f4e 100644 --- a/aw-server/dlp-policy-engine/policy_schema.py +++ b/aw-server/dlp-policy-engine/policy_schema.py @@ -47,6 +47,11 @@ class PolicyActivateRequest(BaseModel): actor: str | None = Field(default="api") +class PolicyStatusRequest(BaseModel): + actor: str | None = Field(default="api") + comment: str | None = Field(default=None, max_length=2048) + + class PolicyRecord(BaseModel): id: int name: str @@ -64,4 +69,3 @@ class PolicyVersionRecord(BaseModel): checksum: str created_at: datetime created_by: str | None - diff --git a/aw-server/dlp-policy-engine/policy_service.py b/aw-server/dlp-policy-engine/policy_service.py index 555dd43..e9dac58 100644 --- a/aw-server/dlp-policy-engine/policy_service.py +++ b/aw-server/dlp-policy-engine/policy_service.py @@ -8,7 +8,7 @@ from typing import Any from fastapi import FastAPI, HTTPException from policy_distributor import build_policy_bundle -from policy_schema import PolicyActivateRequest, PolicyCreateRequest, PolicyUpdateRequest +from policy_schema import PolicyActivateRequest, PolicyCreateRequest, PolicyStatusRequest, PolicyUpdateRequest from policy_storage import PolicyStorage @@ -157,7 +157,34 @@ def update_policy(policy_id: int, payload: PolicyUpdateRequest) -> dict[str, obj @app.post("/api/0/dlp/policies/{policy_id}/activate") def activate_policy(policy_id: int, payload: PolicyActivateRequest) -> dict[str, object]: - item = storage.activate_policy(policy_id=policy_id, actor=payload.actor) + try: + item = storage.activate_policy(policy_id=policy_id, actor=payload.actor) + except Exception as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + if not item: + raise HTTPException(status_code=404, detail="policy not found") + return {"item": item} + + +@app.post("/api/0/dlp/policies/{policy_id}/submit") +def submit_policy_for_approval(policy_id: int, payload: PolicyStatusRequest) -> dict[str, object]: + item = storage.set_policy_status(policy_id, "pending_approval", payload.actor, payload.comment) + if not item: + raise HTTPException(status_code=404, detail="policy not found") + return {"item": item} + + +@app.post("/api/0/dlp/policies/{policy_id}/approve") +def approve_policy(policy_id: int, payload: PolicyStatusRequest) -> dict[str, object]: + item = storage.set_policy_status(policy_id, "approved", payload.actor, payload.comment) + if not item: + raise HTTPException(status_code=404, detail="policy not found") + return {"item": item} + + +@app.post("/api/0/dlp/policies/{policy_id}/draft") +def return_policy_to_draft(policy_id: int, payload: PolicyStatusRequest) -> dict[str, object]: + item = storage.set_policy_status(policy_id, "draft", payload.actor, payload.comment) if not item: raise HTTPException(status_code=404, detail="policy not found") return {"item": item} @@ -173,3 +200,15 @@ def delete_policy(policy_id: int) -> dict[str, bool]: if not deleted: raise HTTPException(status_code=404, detail="policy not found") return {"deleted": True} + + +@app.get("/api/0/dlp/policies/audit") +def list_policy_audit(limit: int = 200) -> dict[str, object]: + return {"items": storage.list_audit(policy_id=None, limit=max(1, min(limit, 1000)))} + + +@app.get("/api/0/dlp/policies/{policy_id}/audit") +def list_single_policy_audit(policy_id: int, limit: int = 200) -> dict[str, object]: + if not storage.get_policy(policy_id): + raise HTTPException(status_code=404, detail="policy not found") + return {"items": storage.list_audit(policy_id=policy_id, limit=max(1, min(limit, 1000)))} diff --git a/aw-server/dlp-policy-engine/policy_storage.py b/aw-server/dlp-policy-engine/policy_storage.py index 6acb6de..3365579 100644 --- a/aw-server/dlp-policy-engine/policy_storage.py +++ b/aw-server/dlp-policy-engine/policy_storage.py @@ -47,6 +47,7 @@ class PolicyStorage: id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, description TEXT, + status TEXT NOT NULL DEFAULT 'draft', is_active INTEGER NOT NULL DEFAULT 0, current_version INTEGER NOT NULL DEFAULT 1, checksum TEXT NOT NULL, @@ -67,16 +68,55 @@ class PolicyStorage: UNIQUE(policy_id, version) ); + CREATE TABLE IF NOT EXISTS policy_audit ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + policy_id INTEGER, + action TEXT NOT NULL, + actor TEXT, + comment TEXT, + details_json TEXT, + created_at TEXT NOT NULL, + FOREIGN KEY(policy_id) REFERENCES policies(id) + ); + CREATE INDEX IF NOT EXISTS idx_policies_active ON policies(is_active); CREATE INDEX IF NOT EXISTS idx_policy_versions_policy ON policy_versions(policy_id, version DESC); + CREATE INDEX IF NOT EXISTS idx_policy_audit_policy ON policy_audit(policy_id, id DESC); """ ) + cols = [r["name"] for r in conn.execute("PRAGMA table_info(policies)").fetchall()] + if "status" not in cols: + conn.execute("ALTER TABLE policies ADD COLUMN status TEXT NOT NULL DEFAULT 'draft'") + + def _audit( + self, + conn: sqlite3.Connection, + policy_id: int | None, + action: str, + actor: str | None, + comment: str | None = None, + details: dict[str, Any] | None = None, + ) -> None: + conn.execute( + """ + INSERT INTO policy_audit(policy_id, action, actor, comment, details_json, created_at) + VALUES(?, ?, ?, ?, ?, ?) + """, + ( + policy_id, + action, + actor, + comment, + canonical_policy_json(details) if details is not None else None, + utc_now(), + ), + ) def list_policies(self) -> list[dict[str, Any]]: with self.connect() as conn: rows = conn.execute( """ - SELECT id, name, description, is_active, current_version, checksum, created_at, updated_at + SELECT id, name, description, status, is_active, current_version, checksum, created_at, updated_at FROM policies ORDER BY is_active DESC, updated_at DESC, id DESC """ @@ -87,7 +127,7 @@ class PolicyStorage: with self.connect() as conn: policy_row = conn.execute( """ - SELECT id, name, description, is_active, current_version, checksum, created_at, updated_at + SELECT id, name, description, status, is_active, current_version, checksum, created_at, updated_at FROM policies WHERE id = ? """, @@ -123,16 +163,17 @@ class PolicyStorage: def create_policy(self, name: str, description: str | None, policy: dict[str, Any], activate: bool, actor: str | None) -> dict[str, Any]: checksum = checksum_policy(policy) now = utc_now() + status = "deployed" if activate else "draft" policy_json = canonical_policy_json(policy) with self.connect() as conn: if activate: conn.execute("UPDATE policies SET is_active = 0") cursor = conn.execute( """ - INSERT INTO policies(name, description, is_active, current_version, checksum, created_at, updated_at) - VALUES(?, ?, ?, 1, ?, ?, ?) + INSERT INTO policies(name, description, status, is_active, current_version, checksum, created_at, updated_at) + VALUES(?, ?, ?, ?, 1, ?, ?, ?) """, - (name, description, 1 if activate else 0, checksum, now, now), + (name, description, status, 1 if activate else 0, checksum, now, now), ) policy_id = int(cursor.lastrowid) conn.execute( @@ -142,6 +183,7 @@ class PolicyStorage: """, (policy_id, policy_json, checksum, now, actor), ) + self._audit(conn, policy_id, "create", actor, details={"activate": activate, "status": status}) return self.get_policy(policy_id) # type: ignore[return-value] def update_policy( @@ -162,10 +204,12 @@ class PolicyStorage: new_description = description if description is not None else current["description"] new_version = int(current["current_version"]) new_checksum = current["checksum"] + new_status = current.get("status", "draft") if policy is not None: new_version += 1 new_checksum = checksum_policy(policy) + new_status = "draft" policy_json = canonical_policy_json(policy) conn.execute( """ @@ -177,16 +221,18 @@ class PolicyStorage: if activate: conn.execute("UPDATE policies SET is_active = 0") + new_status = "deployed" conn.execute( """ UPDATE policies - SET name = ?, description = ?, is_active = ?, current_version = ?, checksum = ?, updated_at = ? + SET name = ?, description = ?, status = ?, is_active = ?, current_version = ?, checksum = ?, updated_at = ? WHERE id = ? """, ( new_name, new_description, + new_status, 1 if activate else current["is_active"], new_version, new_checksum, @@ -194,19 +240,23 @@ class PolicyStorage: policy_id, ), ) + self._audit(conn, policy_id, "update", actor, details={"activate": activate, "status": new_status}) return self.get_policy(policy_id) def activate_policy(self, policy_id: int, actor: str | None) -> dict[str, Any] | None: current = self.get_policy(policy_id) if not current: return None + if current.get("status") != "approved": + raise ValueError("policy must be approved before deploy") with self.connect() as conn: conn.execute("UPDATE policies SET is_active = 0") conn.execute( - "UPDATE policies SET is_active = 1, updated_at = ? WHERE id = ?", + "UPDATE policies SET status = 'deployed', is_active = 1, updated_at = ? WHERE id = ?", (utc_now(), policy_id), ) + self._audit(conn, policy_id, "deploy", actor) return self.get_policy(policy_id) def rollback_active_policy(self, actor: str | None) -> dict[str, Any] | None: @@ -252,11 +302,12 @@ class PolicyStorage: conn.execute( """ UPDATE policies - SET current_version = ?, checksum = ?, updated_at = ? + SET status = 'draft', current_version = ?, checksum = ?, updated_at = ? WHERE id = ? """, (rollback_version, rollback_checksum, now, active["id"]), ) + self._audit(conn, int(active["id"]), "rollback", actor, details={"rollback_to": previous_version}) return self.get_policy(int(active["id"])) def delete_policy(self, policy_id: int) -> bool: @@ -267,6 +318,50 @@ class PolicyStorage: raise ValueError("cannot delete active policy") with self.connect() as conn: + self._audit(conn, policy_id, "delete", None) conn.execute("DELETE FROM policy_versions WHERE policy_id = ?", (policy_id,)) conn.execute("DELETE FROM policies WHERE id = ?", (policy_id,)) return True + + def set_policy_status(self, policy_id: int, status: str, actor: str | None, comment: str | None = None) -> dict[str, Any] | None: + current = self.get_policy(policy_id) + if not current: + return None + allowed = {"draft", "pending_approval", "approved", "deployed"} + if status not in allowed: + raise ValueError(f"unsupported status: {status}") + with self.connect() as conn: + conn.execute( + "UPDATE policies SET status = ?, updated_at = ? WHERE id = ?", + (status, utc_now(), policy_id), + ) + self._audit(conn, policy_id, "status_change", actor, comment=comment, details={"status": status}) + return self.get_policy(policy_id) + + def list_audit(self, policy_id: int | None = None, limit: int = 200) -> list[dict[str, Any]]: + query = """ + SELECT id, policy_id, action, actor, comment, details_json, created_at + FROM policy_audit + """ + params: tuple[Any, ...] + if policy_id is None: + query += " ORDER BY id DESC LIMIT ?" + params = (limit,) + else: + query += " WHERE policy_id = ? ORDER BY id DESC LIMIT ?" + params = (policy_id, limit) + with self.connect() as conn: + rows = conn.execute(query, params).fetchall() + items: list[dict[str, Any]] = [] + for row in rows: + d = dict(row) + if d.get("details_json"): + try: + d["details"] = json.loads(d["details_json"]) + except Exception: + d["details"] = None + else: + d["details"] = None + d.pop("details_json", None) + items.append(d) + return items diff --git a/docs/dlp-policy-engine.md b/docs/dlp-policy-engine.md index d6513e7..9d87ba9 100644 --- a/docs/dlp-policy-engine.md +++ b/docs/dlp-policy-engine.md @@ -2,88 +2,74 @@ ## Purpose -`aw-server/dlp-policy-engine` centralizes DLP policy lifecycle for `AWatch-rus` Windows endpoints. +`aw-server/dlp-policy-engine` is the centralized policy lifecycle service for `AWatch-rus` endpoints. -It does not replace endpoint-local safety. Endpoints can run in: -- `local` -- `server` -- `cached` fallback after server outage +It supports: +- full CRUD for policy documents; +- policy versioning in SQLite; +- approval workflow (`draft -> pending_approval -> approved -> deployed`); +- audit trail for every policy mutation; +- agent push/pull coordination (`heartbeat` + `desired` refresh hint). -## API - -Base URL: +## Base URL ```text http://:5601 ``` -Routes: +## REST API +Health: - `GET /healthz` + +CRUD: - `GET /api/0/dlp/policies` - `POST /api/0/dlp/policies` -- `GET /api/0/dlp/policies/active` -- `POST /api/0/dlp/policies/rollback` - `GET /api/0/dlp/policies/{id}` - `PUT /api/0/dlp/policies/{id}` -- `POST /api/0/dlp/policies/{id}/activate` - `DELETE /api/0/dlp/policies/{id}` -## Policy create example +Active policy: +- `GET /api/0/dlp/policies/active` +- `GET /api/0/dlp/policies/active/version` +- `POST /api/0/dlp/policies/rollback` -```json -{ - "name": "base-windows-policy", - "description": "Primary DLP policy for pilot endpoints", - "activate": true, - "actor": "ansible", - "policy": { - "version": 1, - "defaults": { - "enabled": true, - "cooldownSeconds": 300, - "action": "alert", - "severity": "medium" - }, - "endpoint": { - "clipboard": [], - "usb": [], - "print": [] - } - } -} +Approval workflow: +- `POST /api/0/dlp/policies/{id}/submit` -> `pending_approval` +- `POST /api/0/dlp/policies/{id}/approve` -> `approved` +- `POST /api/0/dlp/policies/{id}/draft` -> `draft` +- `POST /api/0/dlp/policies/{id}/activate` -> deploy (allowed only from `approved`) + +Audit: +- `GET /api/0/dlp/policies/audit?limit=200` +- `GET /api/0/dlp/policies/{id}/audit?limit=200` + +Agent push/pull sync: +- `POST /api/0/dlp/policies/agents/{agent_id}/heartbeat` +- `GET /api/0/dlp/policies/agents/{agent_id}/desired` + +## Workflow Example + +1. Create draft: +`POST /api/0/dlp/policies` +2. Submit: +`POST /api/0/dlp/policies/{id}/submit` +3. Approve: +`POST /api/0/dlp/policies/{id}/approve` +4. Deploy: +`POST /api/0/dlp/policies/{id}/activate` + +Every step is written to `policy_audit`. + +## Deployment + +- Service unit: `aw-dlp-policy-engine.service` +- DB path: `AW_DLP_POLICY_ENGINE_DB_PATH` +- Port: `AW_DLP_POLICY_ENGINE_PORT` (default `5601`) +- Ansible role: `ansible/roles/dlp-policy-engine/tasks/main.yml` + +Recommended server deploy: + +```bash +ansible-playbook -i ansible/inventory.ini ansible/deploy_aw_server.yml ``` - -## Active policy response - -```json -{ - "active": true, - "policyId": 1, - "name": "base-windows-policy", - "version": 3, - "checksum": "sha256...", - "updatedAtUtc": "2026-05-11T12:00:00Z", - "policy": { - "version": 1, - "defaults": { - "enabled": true, - "cooldownSeconds": 300, - "action": "alert", - "severity": "medium" - }, - "endpoint": { - "clipboard": [], - "usb": [], - "print": [] - } - } -} -``` - -## Deployment Notes - -- Service runs as `aw-dlp-policy-engine.service`. -- SQLite path is controlled by `AW_DLP_POLICY_ENGINE_DB_PATH`. -- Default port is `5601`. -- Endpoints should use `server` mode only after `GET /healthz` and `GET /api/0/dlp/policies/active` are confirmed.