From 292b32c173aa8533208dc2f6aeabb1fc551f2fda Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 09:52:58 -0400 Subject: [PATCH 01/29] fix: apply device budgets during auth recovery and revisit stale reviews --- backend/app/api/deps.py | 18 ++++++++------- backend/app/workers/guardian_review_worker.py | 16 ++++++++------ backend/tests/test_device_auth.py | 22 +++++++++++++++++++ 3 files changed, 41 insertions(+), 15 deletions(-) diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index af8116f..590f41b 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -153,6 +153,7 @@ def current_device(request: Request, db: Session = Depends(get_db_dep)) -> Devic raise HTTPException(status_code=401, detail="Empty token") blocked, retry_after = rate_limit.is_blocked("device_auth", client_ip) + device = None if blocked: # A child can legitimately recover from stale bad credentials by # re-pairing. Let a valid structured token clear its own IP's failure @@ -161,14 +162,15 @@ def current_device(request: Request, db: Session = Depends(get_db_dep)) -> Devic device = device_tokens.authenticate(db, token) if device is not None: rate_limit.reset("device_auth", client_ip) - return device - raise HTTPException( - status_code=429, - detail="Too many failed device authentication attempts.", - headers={"Retry-After": str(retry_after)}, - ) - - device = device_tokens.authenticate(db, token) + if device is None: + raise HTTPException( + status_code=429, + detail="Too many failed device authentication attempts.", + headers={"Retry-After": str(retry_after)}, + ) + + if device is None: + device = device_tokens.authenticate(db, token) if device is not None: rate_limit.reset("device_auth", client_ip) operation = device_budget.operation_for_path(request.url.path, request.method) diff --git a/backend/app/workers/guardian_review_worker.py b/backend/app/workers/guardian_review_worker.py index 3fdb788..0b258f1 100644 --- a/backend/app/workers/guardian_review_worker.py +++ b/backend/app/workers/guardian_review_worker.py @@ -20,14 +20,16 @@ async def run_once() -> bool: async def loop() -> None: - db = get_sessionmaker()() - try: - recovered = guardian_review.recover_stale_jobs(db) - if recovered: - log.warning("requeued %d stale Guardian Review job(s)", recovered) - finally: - db.close() while True: + # A job started shortly before a restart may not yet be stale during + # startup. Recheck on every pass so it cannot remain running forever. + db = get_sessionmaker()() + try: + recovered = guardian_review.recover_stale_jobs(db) + if recovered: + log.warning("recovered %d stale Guardian Review job(s)", recovered) + finally: + db.close() try: processed = await run_once() except Exception as exc: # pragma: no cover - supervisor handles repeated failures diff --git a/backend/tests/test_device_auth.py b/backend/tests/test_device_auth.py index bfc0f70..d2610bd 100644 --- a/backend/tests/test_device_auth.py +++ b/backend/tests/test_device_auth.py @@ -196,3 +196,25 @@ def test_valid_device_heartbeat_has_per_device_burst_budget(monkeypatch, tmp_pat ) assert limited.status_code == 429 assert "Retry-After" in limited.headers + + +def test_valid_token_during_ip_block_still_obeys_device_budget(monkeypatch, tmp_path): + client = _client(monkeypatch, tmp_path) + from app.services.device_bootstrap_token import ensure_device_bootstrap_token + + paired = client.post( + "/api/devices/bootstrap-local", + json={"hostname": "kid-pc", "device_bootstrap_token": ensure_device_bootstrap_token()}, + ) + token = paired.json()["device_token"] + path = "/api/devices/heartbeat" + payload = {"queued_frames": 0} + for _ in range(device_budget.BUDGETS["heartbeat"].request_burst): + assert client.post(path, json=payload, headers={"Authorization": f"Bearer {token}"}).status_code == 200 + + for _ in range(rate_limit.MAX_FAILURES): + assert client.post(path, json=payload, headers={"Authorization": "Bearer gn_dev_bogus_bogus"}).status_code == 401 + + recovered = client.post(path, json=payload, headers={"Authorization": f"Bearer {token}"}) + assert recovered.status_code == 429 + assert recovered.json()["detail"] == "Device request budget exceeded." From 00933cbd075cacc84a3ea07ed89a39de1c88b954 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 09:55:02 -0400 Subject: [PATCH 02/29] fix: preserve deletion scope and sanitize recovery snapshots --- backend/app/archive/format.py | 90 ++++++++++++++++++++++++++++ backend/app/db/maintenance.py | 11 +++- backend/app/services/purge.py | 33 ++++++++-- backend/app/services/retention.py | 14 +---- backend/tests/test_archive_format.py | 19 ++++++ backend/tests/test_db_session.py | 19 ++++++ backend/tests/test_purge_cascade.py | 90 +++++++++++++++++++++++++++- 7 files changed, 259 insertions(+), 17 deletions(-) diff --git a/backend/app/archive/format.py b/backend/app/archive/format.py index 8f28403..9081586 100644 --- a/backend/app/archive/format.py +++ b/backend/app/archive/format.py @@ -92,6 +92,90 @@ def _encoded_field(table: str, record_id: dict[str, Any], name: str, value: Any) return encoded +def _sanitize_snapshot(database: Path) -> None: + """Apply the same deletion boundary to the restorable DB and JSONL view. + + This operates only on the copied SQLite backup, never on the live database. + The archive must not reintroduce a tombstoned event or a pending blob whose + file is deliberately excluded from evidence coverage. + """ + with closing(sqlite3.connect(database)) as conn: + conn.execute("PRAGMA foreign_keys=ON") + tables = { + str(row[0]) for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + if not {"events", "evidence_blobs"}.issubset(tables): + return + conn.execute( + "CREATE TEMP TABLE archive_deleted_events AS " + "SELECT event_id FROM events WHERE deletion_state != 'active'" + ) + if "risk_results" in tables: + conn.execute( + "CREATE TEMP TABLE archive_deleted_risks AS " + "SELECT risk_id FROM risk_results WHERE event_id IN " + "(SELECT event_id FROM archive_deleted_events)" + ) + if "alerts" in tables: + conn.execute( + "CREATE TEMP TABLE archive_deleted_alerts AS " + "SELECT alert_id FROM alerts WHERE risk_id IN " + "(SELECT risk_id FROM archive_deleted_risks)" + ) + if "guardian_reviews" in tables: + if "guardian_review_feedback" in tables: + conn.execute( + "DELETE FROM guardian_review_feedback WHERE review_id IN " + "(SELECT review_id FROM guardian_reviews WHERE alert_id IN " + "(SELECT alert_id FROM archive_deleted_alerts))" + ) + conn.execute( + "DELETE FROM guardian_reviews WHERE alert_id IN " + "(SELECT alert_id FROM archive_deleted_alerts)" + ) + if "guardian_review_previews" in tables: + conn.execute( + "DELETE FROM guardian_review_previews WHERE alert_id IN " + "(SELECT alert_id FROM archive_deleted_alerts)" + ) + if "notification_jobs" in tables: + conn.execute( + "DELETE FROM notification_jobs WHERE alert_id IN " + "(SELECT alert_id FROM archive_deleted_alerts)" + ) + conn.execute( + "DELETE FROM alerts WHERE alert_id IN " + "(SELECT alert_id FROM archive_deleted_alerts)" + ) + if "risk_results" in tables: + conn.execute( + "DELETE FROM risk_results WHERE risk_id IN " + "(SELECT risk_id FROM archive_deleted_risks)" + ) + conn.execute( + "UPDATE events SET screenshot_blob_id=NULL WHERE screenshot_blob_id IN " + "(SELECT blob_id FROM evidence_blobs WHERE deletion_state != 'active')" + ) + conn.execute( + "UPDATE events SET image_blob_id=NULL WHERE image_blob_id IN " + "(SELECT blob_id FROM evidence_blobs WHERE deletion_state != 'active')" + ) + conn.execute( + "DELETE FROM evidence_blobs WHERE deletion_state != 'active' OR event_id IN " + "(SELECT event_id FROM archive_deleted_events)" + ) + conn.execute( + "DELETE FROM events WHERE event_id IN " + "(SELECT event_id FROM archive_deleted_events)" + ) + violations = conn.execute("PRAGMA foreign_key_check").fetchall() + if violations: + raise ArchiveError("archive snapshot contains broken references") + conn.commit() + + def _write_logical_records(database: Path, records_dir: Path) -> dict[str, int]: counts: dict[str, int] = {} records_dir.mkdir(parents=True, exist_ok=True) @@ -222,6 +306,7 @@ def create_archive( # The database backup helper emits an adjacent implementation manifest; # GNA has its own complete signed manifest, so do not duplicate it. database.with_name(database.name + ".manifest.json").unlink(missing_ok=True) + _sanitize_snapshot(database) counts = _write_logical_records(database, payload / "records") evidence_count, evidence_bytes, evidence_map = _copy_evidence(data_dir, payload, database) _write_json(payload / "evidence-map.json", evidence_map) @@ -624,6 +709,11 @@ def restore_archive( ) conn.execute("UPDATE pairing_codes SET used=1") conn.execute("UPDATE notification_jobs SET status='cancelled'") + conn.execute( + "UPDATE guardian_reviews SET status='cancelled', " + "error_code='restored_reauthorization_required' " + "WHERE status IN ('queued', 'running')" + ) for key in ("notification_settings", "complete_backup_config"): row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone() if row: diff --git a/backend/app/db/maintenance.py b/backend/app/db/maintenance.py index b5d75e8..a0a6e9a 100644 --- a/backend/app/db/maintenance.py +++ b/backend/app/db/maintenance.py @@ -30,7 +30,16 @@ class IntegrityResult: BACKUP_MANIFEST_FORMAT = "guardiannode-sqlite-backup-v1" -SUPPORTED_SCHEMA_REVISIONS = {None, "0001_beta_baseline", "0002_complete_backups"} +SUPPORTED_SCHEMA_REVISIONS = { + None, + "0001_beta_baseline", + "0002_complete_backups", + "0003_guardian_reviews", + "0004_guardian_review_privacy", + "0005_guardian_review_feedback", + "0006_beta_security_break", + "0007_family_beta_workflows", +} def backup_manifest_path(database: Path) -> Path: diff --git a/backend/app/services/purge.py b/backend/app/services/purge.py index aa3c030..f4c3257 100644 --- a/backend/app/services/purge.py +++ b/backend/app/services/purge.py @@ -1,7 +1,7 @@ """Cascade-safe deletion of events and their dependent records. Events fan out into RiskResults, Alerts, EvidenceBlob rows, and encrypted blob -files on disk. SQLite (as configured) does not enforce FK cascades, so every +files on disk. SQLite enforces foreign keys but does not cascade them, so every delete path — retention cleanup and parent-initiated wipes — must go through these helpers or it strands orphaned rows and undeletable evidence files. """ @@ -18,7 +18,9 @@ Event, EvidenceBlob, GuardianReview, + GuardianReviewFeedback, GuardianReviewPreview, + NotificationJob, RiskResult, ) from app.services.evidence_paths import UnsafeEvidencePathError, resolve_stored_evidence_path @@ -65,6 +67,9 @@ def delete_events(session: Session, event_ids: Iterable[str]) -> dict[str, int]: rows = session.query(Event).filter(Event.event_id.in_(chunk)).all() deletable_ids: list[str] = [] for event in rows: + event.deletion_state = "pending_delete" + event.deletion_requested_at = event.deletion_requested_at or datetime.now(UTC) + session.flush() blobs = session.query(EvidenceBlob).filter(EvidenceBlob.event_id == event.event_id).all() failed = False for blob in blobs: @@ -95,6 +100,17 @@ def delete_events(session: Session, event_ids: Iterable[str]) -> dict[str, int]: row[0] for row in session.query(GuardianReviewPreview.preview_id) .filter(GuardianReviewPreview.alert_id.in_(alert_ids)).all() ] + review_ids = [ + row[0] for row in session.query(GuardianReview.review_id) + .filter(GuardianReview.alert_id.in_(alert_ids)).all() + ] + if review_ids: + session.query(GuardianReviewFeedback).filter( + GuardianReviewFeedback.review_id.in_(review_ids) + ).delete(synchronize_session=False) + session.query(NotificationJob).filter( + NotificationJob.alert_id.in_(alert_ids) + ).delete(synchronize_session=False) deleted["guardian_reviews"] += session.query(GuardianReview).filter( GuardianReview.alert_id.in_(alert_ids) ).delete(synchronize_session=False) @@ -131,7 +147,7 @@ def delete_orphaned_blob_files(session: Session) -> int: def retry_pending_deletions(session: Session, *, limit: int = 100) -> dict[str, int]: - """Retry failed blob unlinks and finish their pending event cascades.""" + """Retry file deletion, preserving events with only expired blobs.""" blobs = ( session.query(EvidenceBlob) .filter(EvidenceBlob.deletion_state.in_(["pending_delete", "delete_failed"])) @@ -140,11 +156,20 @@ def retry_pending_deletions(session: Session, *, limit: int = 100) -> dict[str, .all() ) event_ids = {blob.event_id for blob in blobs if blob.event_id} - removed = sum(1 for blob in blobs if delete_blob(session, blob)) + removed = 0 + for blob in blobs: + if delete_blob(session, blob): + removed += 1 + for event in session.query(Event).filter(Event.screenshot_blob_id == blob.blob_id).all(): + event.screenshot_blob_id = None + for event in session.query(Event).filter(Event.image_blob_id == blob.blob_id).all(): + event.image_blob_id = None session.flush() finished = {"events": 0, "risk_results": 0, "alerts": 0, "blobs": removed, "guardian_reviews": 0} for event_id in event_ids: - if session.query(EvidenceBlob).filter(EvidenceBlob.event_id == event_id).count() == 0: + event = session.get(Event, event_id) + if (event is not None and event.deletion_state == "pending_delete" + and session.query(EvidenceBlob).filter(EvidenceBlob.event_id == event_id).count() == 0): counts = delete_events(session, [event_id]) for key, value in counts.items(): finished[key] += value diff --git a/backend/app/services/retention.py b/backend/app/services/retention.py index 8f7f1b8..0df9af2 100644 --- a/backend/app/services/retention.py +++ b/backend/app/services/retention.py @@ -6,7 +6,7 @@ from sqlalchemy.orm import Session -from app.db.models import Alert, AuditLog, Event, EvidenceBlob, RiskResult +from app.db.models import AuditLog, Event, EvidenceBlob, RiskResult log = logging.getLogger(__name__) @@ -46,22 +46,14 @@ def run_cleanup(session: Session, retention: dict[str, int] | None = None) -> di event_ids = [ row[0] for row in session.query(RiskResult.event_id) - .filter(RiskResult.risk_level.in_(levels), RiskResult.created_at < cutoff) + .join(Event, RiskResult.event_id == Event.event_id) + .filter(RiskResult.risk_level.in_(levels), Event.timestamp < cutoff) .all() ] counts = purge.delete_events(session, event_ids) for key in ("alerts", "risk_results", "events", "blobs"): deleted[key] += counts[key] - # Defensive: also expire alert rows of this severity whose chain is - # already gone (e.g. rows from before cascade deletion existed). - sev_levels = [sev_group] - deleted["alerts"] += ( - session.query(Alert) - .filter(Alert.severity.in_(sev_levels), Alert.created_at < cutoff) - .delete(synchronize_session=False) - ) - # 2) Events that never got a risk result (ingest interrupted mid-pipeline). cutoff = _cutoff(max(r["low"], 1)) orphan_event_ids = [ diff --git a/backend/tests/test_archive_format.py b/backend/tests/test_archive_format.py index 7e8336c..39f6f9b 100644 --- a/backend/tests/test_archive_format.py +++ b/backend/tests/test_archive_format.py @@ -22,6 +22,7 @@ _safe_member_target, _safe_members, create_archive, + extract_archive, inspect_archive, restore_archive, verify_archive, @@ -111,6 +112,24 @@ def test_passphrase_archive_round_trip_and_restore(archive_instance, tmp_path: P assert (target / "keys" / "master.key").stat().st_size == 32 +def test_pending_deletion_excluded_from_both_archive_views(archive_instance, tmp_path: Path): + data_dir, database = archive_instance + with sqlite3.connect(database) as conn: + conn.execute("UPDATE events SET deletion_state='pending_delete' WHERE event_id='event-1'") + archive = tmp_path / "pending.gna" + create_archive( + archive, data_dir=data_dir, db_url=f"sqlite:///{database}", + passphrase="synthetic-test-passphrase", + ) + extracted = tmp_path / "extracted" + extract_archive(archive, extracted, passphrase="synthetic-test-passphrase") + assert not (extracted / "records" / "events.jsonl").read_text().strip() + with sqlite3.connect(extracted / "database.sqlite3") as conn: + assert conn.execute("SELECT COUNT(*) FROM events WHERE event_id='event-1'").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM evidence_blobs WHERE blob_id='blob'").fetchone()[0] == 0 + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] + + def test_recipient_archive_and_wrong_key_fail(archive_instance, tmp_path: Path): data_dir, database = archive_instance private_path = tmp_path / "recovery.pem" diff --git a/backend/tests/test_db_session.py b/backend/tests/test_db_session.py index b9138d4..52ea63b 100644 --- a/backend/tests/test_db_session.py +++ b/backend/tests/test_db_session.py @@ -104,3 +104,22 @@ def test_backup_refuses_to_overwrite_without_flag(tmp_path): with pytest.raises(Exception, match="already exists"): backup_database(destination, source=source) + + +def test_current_schema_backup_restores(tmp_path, monkeypatch): + from app import settings as settings_mod + from app.db.maintenance import database_schema_revision + from app.db.migrations import upgrade_schema + from app.db.session import configure_sqlite_engine + + source = tmp_path / "current.sqlite" + monkeypatch.setattr(settings_mod.settings, "db_url", f"sqlite:///{source}") + engine = configure_sqlite_engine(create_engine(f"sqlite:///{source}")) + upgrade_schema(engine) + engine.dispose() + assert database_schema_revision(source) == "0007_family_beta_workflows" + backup = tmp_path / "current-backup.sqlite" + backup_database(backup, source=source) + restored = tmp_path / "current-restored.sqlite" + restore_database(backup, destination=restored) + assert database_schema_revision(restored) == "0007_family_beta_workflows" diff --git a/backend/tests/test_purge_cascade.py b/backend/tests/test_purge_cascade.py index 68f1d47..d8f6f1a 100644 --- a/backend/tests/test_purge_cascade.py +++ b/backend/tests/test_purge_cascade.py @@ -4,7 +4,18 @@ from datetime import UTC, datetime, timedelta from app import settings as settings_mod -from app.db.models import Alert, Device, Event, EvidenceBlob, RiskResult +from app.db.models import ( + Alert, + Device, + Event, + EvidenceBlob, + GuardianReview, + GuardianReviewFeedback, + GuardianReviewPreview, + NotificationJob, + RiskResult, + User, +) from app.services import purge, retention @@ -68,6 +79,83 @@ def test_delete_events_cascades_risk_alert_blob(db_session, tmp_path): assert not blob_path.exists() +def test_delete_events_removes_review_feedback_and_notification_jobs(db_session, tmp_path): + s = db_session + event_id, _, _, _ = _mk_chain(s, tmp_path, suffix="dependent", severity="high", age_days=0) + user = User(display_name="Synthetic Parent", password_hash="x", recovery_hash="x") + s.add(user) + s.flush() + s.add(GuardianReviewPreview( + preview_id="preview-dependent", alert_id="a-dependent", actor_user_id=user.id, + provider="mock", model_requested="mock", schema_version="1", prompt_version="v1", + payload_digest="0" * 64, incident_fingerprint="1" * 64, + expires_at=datetime.now(UTC) + timedelta(minutes=10), + )) + s.flush() + s.add(GuardianReview( + review_id="review-dependent", preview_id="preview-dependent", alert_id="a-dependent", + requester_user_id=user.id, provider="mock", dedup_key="2" * 64, + schema_version="1", prompt_version="v1", model_requested="mock", status="completed", + )) + s.flush() + s.add(GuardianReviewFeedback( + feedback_id="feedback-dependent", review_id="review-dependent", user_id=user.id, + labels=["helpful"], schema_version="1", prompt_version="v1", + redaction_version="v1", model="mock", + )) + s.add(NotificationJob(alert_id="a-dependent", severity="high", risk_summary="synthetic")) + s.commit() + + purge.delete_events(s, [event_id]) + s.commit() + assert s.get(Event, event_id) is None + assert s.query(GuardianReviewFeedback).count() == 0 + assert s.query(NotificationJob).count() == 0 + assert s.connection().exec_driver_sql("PRAGMA foreign_key_check").fetchall() == [] + + +def test_retry_expired_blob_keeps_unexpired_high_alert(db_session, tmp_path, monkeypatch): + s = db_session + event_id, _, blob_id, path = _mk_chain( + s, tmp_path, suffix="blob-retry", severity="high", age_days=40, + ) + original_unlink = type(path).unlink + + def locked(candidate, *args, **kwargs): + if candidate == path: + raise PermissionError("synthetic lock") + return original_unlink(candidate, *args, **kwargs) + + monkeypatch.setattr(type(path), "unlink", locked) + retention.run_cleanup(s) + assert s.get(EvidenceBlob, blob_id).deletion_state == "delete_failed" + monkeypatch.setattr(type(path), "unlink", original_unlink) + purge.retry_pending_deletions(s) + s.commit() + assert s.get(Event, event_id) is not None + assert s.get(Event, event_id).screenshot_blob_id is None + assert s.get(Alert, "a-blob-retry") is not None + assert s.get(EvidenceBlob, blob_id) is None + + +def test_failed_full_retention_keeps_alert_until_retry(db_session, tmp_path, monkeypatch): + s = db_session + event_id, _, _, path = _mk_chain( + s, tmp_path, suffix="full-retry", severity="high", age_days=100, + ) + original_unlink = type(path).unlink + + def locked(candidate, *args, **kwargs): + if candidate == path: + raise PermissionError("synthetic lock") + return original_unlink(candidate, *args, **kwargs) + + monkeypatch.setattr(type(path), "unlink", locked) + retention.run_cleanup(s) + assert s.get(Event, event_id).deletion_state == "pending_delete" + assert s.get(Alert, "a-full-retry") is not None + + def test_wipe_old_events_leaves_no_orphans(db_session, tmp_path): """The storage-wipe path (older_than_days) must clear the full chain.""" s = db_session From 88aaf1408f9189b6310509f235838ecec6998cd2 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 09:57:02 -0400 Subject: [PATCH 03/29] fix: use masked parent step-up dialog for protected actions --- dashboard/e2e/family-dashboard.spec.ts | 36 +++++++++ dashboard/src/App.tsx | 2 + dashboard/src/api.ts | 12 +-- .../src/components/StepUpDialog.test.tsx | 47 +++++++++++ dashboard/src/components/StepUpDialog.tsx | 77 +++++++++++++++++++ dashboard/src/stepUp.ts | 18 +++++ 6 files changed, 182 insertions(+), 10 deletions(-) create mode 100644 dashboard/src/components/StepUpDialog.test.tsx create mode 100644 dashboard/src/components/StepUpDialog.tsx create mode 100644 dashboard/src/stepUp.ts diff --git a/dashboard/e2e/family-dashboard.spec.ts b/dashboard/e2e/family-dashboard.spec.ts index 0603356..656b189 100644 --- a/dashboard/e2e/family-dashboard.spec.ts +++ b/dashboard/e2e/family-dashboard.spec.ts @@ -213,3 +213,39 @@ test("secure pairing instructions expose the exact trust handoff", async ({ page await expect(page.getByText("cedar · harbor · maple · river")).toBeVisible(); await expect(page.getByRole("link", { name: "Download .gnpair bundle" })).toHaveAttribute("download", ""); }); + +test("protected pause uses masked step-up and resumes only after confirmation", async ({ page }, testInfo) => { + const errors = collectBrowserErrors(page); + let pauseCalls = 0; + let confirmed = false; + await page.route("**/api/devices/device-1/pause", async (route) => { + pauseCalls += 1; + if (!confirmed) return fulfill(route, { detail: { code: "step_up_required", level: "standard" } }, 403); + return fulfill(route, { ok: true }); + }); + await page.route("**/api/auth/reauth", async (route) => { + const body = route.request().postDataJSON(); + if (body.password !== "synthetic-browser-password") return fulfill(route, { detail: "Invalid password" }, 401); + confirmed = true; + return fulfill(route, { ok: true }); + }); + + await page.goto("/devices"); + await expect(page).toHaveTitle(/GuardianNode/); + await page.getByRole("button", { name: "Pause", exact: true }).click(); + const dialog = page.getByRole("dialog"); + await expect(dialog.getByRole("heading", { name: "Confirm your password" })).toBeVisible(); + await expect(dialog.getByLabel("Parent password")).toHaveAttribute("type", "password"); + await mkdir("/tmp/guardiannode-qa", { recursive: true }); + await page.screenshot({ path: `/tmp/guardiannode-qa/${testInfo.project.name}-step-up.png`, fullPage: false }); + await dialog.getByRole("button", { name: "Cancel" }).click(); + await expect(dialog).not.toBeVisible(); + expect(pauseCalls).toBe(1); + + await page.getByRole("button", { name: "Pause", exact: true }).click(); + await dialog.getByLabel("Parent password").fill("synthetic-browser-password"); + await dialog.getByRole("button", { name: "Continue" }).click(); + await expect(page.getByText("Monitoring pause recorded.")).toBeVisible(); + expect(pauseCalls).toBe(3); + expect(errors.filter((message) => !message.includes("status of 403 (Forbidden)"))).toEqual([]); +}); diff --git a/dashboard/src/App.tsx b/dashboard/src/App.tsx index 4a30400..3a375c9 100644 --- a/dashboard/src/App.tsx +++ b/dashboard/src/App.tsx @@ -2,6 +2,7 @@ import { lazy, Suspense, useEffect, useState } from "react"; import { Navigate, Route, Routes, useNavigate } from "react-router-dom"; import { api } from "./api"; import Layout from "./components/Layout"; +import StepUpDialog from "./components/StepUpDialog"; import Login from "./pages/Login"; import Setup from "./pages/Setup"; @@ -81,6 +82,7 @@ export default function App() { return ( + Loading page…}> } /> diff --git a/dashboard/src/api.ts b/dashboard/src/api.ts index 34b0385..d6aedb1 100644 --- a/dashboard/src/api.ts +++ b/dashboard/src/api.ts @@ -1,5 +1,6 @@ // Typed API client. Uses cookie auth (no token to manage in localStorage). import type { components } from "./api-schema"; +import { requestStepUp } from "./stepUp"; export type Alert = components["schemas"]["AlertDTO"]; export type AlertPage = components["schemas"]["AlertPage"]; @@ -66,16 +67,7 @@ async function request(path: string, init: RequestInit = {}): Promise { if (res.status === 403 && mutating) { const challenge = await res.clone().json().catch(() => null); if (challenge?.detail?.code === "step_up_required") { - const password = window.prompt( - challenge.detail.level === "critical" - ? "Confirm your password for this critical security action." - : "Confirm your password to continue.", - ); - if (password === null) throw new Error("Recent authentication required"); - await request<{ ok: boolean }>("/auth/reauth", { - method: "POST", - body: JSON.stringify({ password }), - }); + await requestStepUp(challenge.detail.level); } else { cachedCsrfToken = null; headers.set("X-CSRF-Token", await getCsrfToken()); diff --git a/dashboard/src/components/StepUpDialog.test.tsx b/dashboard/src/components/StepUpDialog.test.tsx new file mode 100644 index 0000000..6ff3e3e --- /dev/null +++ b/dashboard/src/components/StepUpDialog.test.tsx @@ -0,0 +1,47 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { act, cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { api } from "../api"; +import { requestStepUp } from "../stepUp"; +import StepUpDialog from "./StepUpDialog"; + +afterEach(() => { + cleanup(); + vi.restoreAllMocks(); +}); + +function mount() { + vi.stubGlobal("HTMLDialogElement", HTMLDialogElement); + HTMLDialogElement.prototype.showModal = vi.fn(); + HTMLDialogElement.prototype.close = vi.fn(); + return render(); +} + +it("masks the password and resumes concurrent actions only after successful reauth", async () => { + mount(); + const reauth = vi.spyOn(api, "reauth").mockResolvedValue({ ok: true }); + let first!: Promise; + let second!: Promise; + await act(async () => { + first = requestStepUp("critical"); + second = requestStepUp("standard"); + }); + expect(first).toBe(second); + const input = screen.getByLabelText("Parent password") as HTMLInputElement; + expect(input.type).toBe("password"); + expect(input.autocomplete).toBe("current-password"); + fireEvent.change(input, { target: { value: "synthetic-password" } }); + fireEvent.click(screen.getByText("Continue")); + await first; + await waitFor(() => expect(reauth).toHaveBeenCalledOnce()); + expect(reauth).toHaveBeenCalledWith("synthetic-password"); +}); + +it("cancels without reauthenticating", async () => { + mount(); + const reauth = vi.spyOn(api, "reauth"); + let pending!: Promise; + await act(async () => { pending = requestStepUp("standard"); }); + fireEvent.click(screen.getByText("Cancel")); + await expect(pending).rejects.toThrow("cancelled"); + expect(reauth).not.toHaveBeenCalled(); +}); diff --git a/dashboard/src/components/StepUpDialog.tsx b/dashboard/src/components/StepUpDialog.tsx new file mode 100644 index 0000000..c5e00c7 --- /dev/null +++ b/dashboard/src/components/StepUpDialog.tsx @@ -0,0 +1,77 @@ +import { useEffect, useRef, useState } from "react"; +import { api, ApiError } from "../api"; +import { registerStepUpHandler } from "../stepUp"; + +type Challenge = { + level: string; + resolve: () => void; + reject: (reason: Error) => void; +}; + +export default function StepUpDialog() { + const [challenge, setChallenge] = useState(null); + const [password, setPassword] = useState(""); + const [error, setError] = useState(""); + const [busy, setBusy] = useState(false); + const dialogRef = useRef(null); + + useEffect(() => registerStepUpHandler((level) => new Promise((resolve, reject) => { + setPassword(""); + setError(""); + setChallenge({ level, resolve, reject }); + })), []); + + useEffect(() => { + if (challenge) dialogRef.current?.showModal(); + else dialogRef.current?.close(); + }, [challenge]); + + function cancel() { + if (busy || !challenge) return; + challenge.reject(new Error("Password confirmation cancelled.")); + setPassword(""); + setChallenge(null); + } + + async function confirm(event: React.FormEvent) { + event.preventDefault(); + if (!challenge || busy) return; + setBusy(true); + setError(""); + try { + await api.reauth(password); + challenge.resolve(); + setPassword(""); + setChallenge(null); + } catch (cause) { + setError(cause instanceof ApiError ? cause.message : "Could not confirm your password. Try again."); + setPassword(""); + } finally { + setBusy(false); + } + } + + return ( + { event.preventDefault(); cancel(); }} + className="step-up-dialog fixed left-1/2 top-1/2 w-[calc(100%-2rem)] max-w-sm -translate-x-1/2 -translate-y-1/2 rounded-xl border border-gray-200 bg-white p-6 text-gray-900 shadow-xl backdrop:bg-gray-950/50"> + {challenge &&
+

Confirm your password

+

+ {challenge.level === "critical" ? "Confirm this security action." : "Confirm this action."} +

+ + setPassword(event.target.value)} + className="rounded-md border border-gray-300 px-3 py-2" /> + {error &&

{error}

} +
+ + +
+
} +
+ ); +} diff --git a/dashboard/src/stepUp.ts b/dashboard/src/stepUp.ts new file mode 100644 index 0000000..e604c79 --- /dev/null +++ b/dashboard/src/stepUp.ts @@ -0,0 +1,18 @@ +type StepUpHandler = (level: string) => Promise; + +let handler: StepUpHandler | null = null; +let pending: Promise | null = null; + +export function registerStepUpHandler(next: StepUpHandler): () => void { + handler = next; + return () => { + if (handler === next) handler = null; + }; +} + +export function requestStepUp(level: string): Promise { + if (pending) return pending; + if (!handler) return Promise.reject(new Error("Sign in again to continue.")); + pending = handler(level).finally(() => { pending = null; }); + return pending; +} From 0d56db842f46cb8d91089de2c7adbf7fd493f22c Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 09:57:35 -0400 Subject: [PATCH 04/29] fix: hide tombstoned incidents from parent read paths --- backend/app/api/alerts.py | 29 ++++++++++++++----- backend/app/api/risks.py | 16 ++++++++-- backend/app/services/alert_dedup.py | 5 +++- backend/app/services/daily_digest.py | 5 +++- .../services/guardian_review_minimization.py | 2 +- backend/app/services/purge.py | 4 +++ backend/tests/test_purge_cascade.py | 24 +++++++++++++++ 7 files changed, 71 insertions(+), 14 deletions(-) diff --git a/backend/app/api/alerts.py b/backend/app/api/alerts.py index 20bd043..4a38aed 100644 --- a/backend/app/api/alerts.py +++ b/backend/app/api/alerts.py @@ -68,6 +68,19 @@ class AlertPage(BaseModel): open_count: int +def _visible_alerts(db: Session): + return ( + db.query(Alert) + .join(RiskResult, Alert.risk_id == RiskResult.risk_id) + .join(Event, RiskResult.event_id == Event.event_id) + .filter(Event.deletion_state == "active") + ) + + +def _visible_alert(db: Session, alert_id: str) -> Alert | None: + return _visible_alerts(db).filter(Alert.alert_id == alert_id).first() + + @router.get("", response_model=AlertPage) def list_alerts( severity: str | None = None, @@ -79,7 +92,7 @@ def list_alerts( db: Session = Depends(get_db_dep), _: User = Depends(current_user), ): - q = db.query(Alert) + q = _visible_alerts(db) if severity: q = q.filter(Alert.severity == severity) if status: @@ -89,7 +102,7 @@ def list_alerts( if profile_id: q = q.filter(Alert.profile_id == profile_id) if cursor: - anchor = db.get(Alert, cursor) + anchor = _visible_alert(db, cursor) if anchor is None: raise HTTPException(400, "Invalid alert cursor") q = q.filter(or_( @@ -118,7 +131,7 @@ def list_alerts( return AlertPage( items=out, next_cursor=rows[-1].alert_id if len(rows_with_extra) > limit else None, - open_count=db.query(Alert).filter(Alert.status == "open").count(), + open_count=_visible_alerts(db).filter(Alert.status == "open").count(), ) @@ -137,7 +150,7 @@ def get_alert( db: Session = Depends(get_db_dep), user: User = Depends(current_user), ): - a = db.get(Alert, alert_id) + a = _visible_alert(db, alert_id) if a is None: raise HTTPException(404, "Alert not found") r = db.get(RiskResult, a.risk_id) @@ -201,7 +214,7 @@ def review_alert( db: Session = Depends(get_db_dep), user: User = Depends(current_user), ): - a = db.get(Alert, alert_id) + a = _visible_alert(db, alert_id) if a is None: raise HTTPException(404, "Alert not found") a.status = req.status @@ -229,7 +242,7 @@ def record_feedback( db: Session = Depends(get_db_dep), user: User = Depends(current_user), ): - a = db.get(Alert, alert_id) + a = _visible_alert(db, alert_id) if a is None: raise HTTPException(404, "Alert not found") r = db.get(RiskResult, a.risk_id) @@ -264,7 +277,7 @@ def take_action( db: Session = Depends(get_db_dep), user: User = Depends(current_user), ): - a = db.get(Alert, alert_id) + a = _visible_alert(db, alert_id) if a is None: raise HTTPException(404, "Alert not found") if req.action in {"pause_app", "block_app", "delete_evidence"}: @@ -385,7 +398,7 @@ def create_confirmed_action( user: User = Depends(current_user), _: None = Depends(require_recent_auth), ): - alert = db.get(Alert, alert_id) + alert = _visible_alert(db, alert_id) if alert is None: raise HTTPException(404, "Alert not found") if not alert.device_id: diff --git a/backend/app/api/risks.py b/backend/app/api/risks.py index b05b5e5..49f07c8 100644 --- a/backend/app/api/risks.py +++ b/backend/app/api/risks.py @@ -8,7 +8,7 @@ from sqlalchemy.orm import Session from app.api.deps import current_user, get_db_dep -from app.db.models import RiskResult, User +from app.db.models import Event, RiskResult, User from app.services.audit import log_action router = APIRouter(prefix="/risks", tags=["risks"]) @@ -47,7 +47,12 @@ def list_risks( db: Session = Depends(get_db_dep), _: User = Depends(current_user), ): - q = db.query(RiskResult).order_by(RiskResult.created_at.desc()) + q = ( + db.query(RiskResult) + .join(Event, RiskResult.event_id == Event.event_id) + .filter(Event.deletion_state == "active") + .order_by(RiskResult.created_at.desc()) + ) if risk_level: q = q.filter(RiskResult.risk_level == risk_level) return [_to_dto(r) for r in q.limit(limit).all()] @@ -65,7 +70,12 @@ def feedback( db: Session = Depends(get_db_dep), user: User = Depends(current_user), ): - r = db.get(RiskResult, risk_id) + r = ( + db.query(RiskResult) + .join(Event, RiskResult.event_id == Event.event_id) + .filter(RiskResult.risk_id == risk_id, Event.deletion_state == "active") + .first() + ) if r is None: raise HTTPException(404, "Risk not found") if req.is_false_positive: diff --git a/backend/app/services/alert_dedup.py b/backend/app/services/alert_dedup.py index 47c554b..ebea234 100644 --- a/backend/app/services/alert_dedup.py +++ b/backend/app/services/alert_dedup.py @@ -20,7 +20,7 @@ from sqlalchemy.orm import Session from ulid import ULID -from app.db.models import Alert +from app.db.models import Alert, Event, RiskResult from app.services.audit import log_action from app.settings import settings @@ -63,7 +63,10 @@ def upsert_alert( cutoff = now - window existing = ( session.query(Alert) + .join(RiskResult, Alert.risk_id == RiskResult.risk_id) + .join(Event, RiskResult.event_id == Event.event_id) .filter( + Event.deletion_state == "active", Alert.dedup_key == key, Alert.status == "open", or_(Alert.last_seen_at >= cutoff, Alert.created_at >= cutoff), diff --git a/backend/app/services/daily_digest.py b/backend/app/services/daily_digest.py index 855e3ac..bd410d4 100644 --- a/backend/app/services/daily_digest.py +++ b/backend/app/services/daily_digest.py @@ -9,7 +9,7 @@ from sqlalchemy.orm import Session from ulid import ULID -from app.db.models import Alert, DigestRun, RiskResult, Setting +from app.db.models import Alert, DigestRun, Event, RiskResult, Setting from app.services import notifications @@ -43,6 +43,9 @@ def build_summary(session: Session, *, local_day: date, timezone: str) -> dict: start, end = _day_bounds(local_day, timezone) alerts = ( session.query(Alert) + .join(RiskResult, Alert.risk_id == RiskResult.risk_id) + .join(Event, RiskResult.event_id == Event.event_id) + .filter(Event.deletion_state == "active") .filter(Alert.created_at >= start, Alert.created_at < end) .order_by(Alert.created_at.desc()) .all() diff --git a/backend/app/services/guardian_review_minimization.py b/backend/app/services/guardian_review_minimization.py index ff985f6..2b40eea 100644 --- a/backend/app/services/guardian_review_minimization.py +++ b/backend/app/services/guardian_review_minimization.py @@ -252,7 +252,7 @@ def build_minimized_incident( alert = db.get(Alert, alert_id) risk = db.get(RiskResult, alert.risk_id) if alert else None event = db.get(Event, risk.event_id) if risk else None - if alert is None or risk is None or event is None: + if alert is None or risk is None or event is None or event.deletion_state != "active": raise InvalidIncidentError("Incident not found or incomplete") profile_id = alert.profile_id or event.profile_id profile = db.get(ChildProfile, profile_id) if profile_id else None diff --git a/backend/app/services/purge.py b/backend/app/services/purge.py index f4c3257..eb8a930 100644 --- a/backend/app/services/purge.py +++ b/backend/app/services/purge.py @@ -34,6 +34,10 @@ def delete_blob(session: Session, blob: EvidenceBlob) -> bool: blob.deletion_state = "pending_delete" blob.delete_requested_at = blob.delete_requested_at or now blob.delete_attempts = int(blob.delete_attempts or 0) + 1 + # Check database constraints before unlinking the only on-disk copy. + # The caller still owns the transaction; durable intent requires a + # separate commit before this function is called. + session.flush() try: resolve_stored_evidence_path(blob.encrypted_path).unlink(missing_ok=True) except UnsafeEvidencePathError as e: diff --git a/backend/tests/test_purge_cascade.py b/backend/tests/test_purge_cascade.py index d8f6f1a..863da57 100644 --- a/backend/tests/test_purge_cascade.py +++ b/backend/tests/test_purge_cascade.py @@ -3,6 +3,9 @@ from datetime import UTC, datetime, timedelta +import pytest +from fastapi import HTTPException + from app import settings as settings_mod from app.db.models import ( Alert, @@ -156,6 +159,27 @@ def locked(candidate, *args, **kwargs): assert s.get(Alert, "a-full-retry") is not None +def test_tombstoned_alert_and_risk_are_hidden_from_read_apis(db_session, tmp_path): + from app.api.alerts import get_alert, list_alerts + from app.api.risks import list_risks + + s = db_session + event_id, _, _, _ = _mk_chain( + s, tmp_path, suffix="hidden", severity="high", age_days=0, + ) + user = User(display_name="Synthetic Parent", password_hash="x", recovery_hash="x") + s.add(user) + event = s.get(Event, event_id) + event.deletion_state = "pending_delete" + s.commit() + + assert list_alerts(db=s, _=user, limit=50).items == [] + assert list_risks(db=s, _=user, limit=100) == [] + with pytest.raises(HTTPException) as exc: + get_alert("a-hidden", db=s, user=user) + assert exc.value.status_code == 404 + + def test_wipe_old_events_leaves_no_orphans(db_session, tmp_path): """The storage-wipe path (older_than_days) must clear the full chain.""" s = db_session From c1441ae89385f2ad4e095e9fabc538d4b6ddb033 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 09:58:03 -0400 Subject: [PATCH 05/29] test: verify SQL failure leaves evidence file intact --- backend/tests/test_purge_cascade.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/backend/tests/test_purge_cascade.py b/backend/tests/test_purge_cascade.py index 863da57..1d53de5 100644 --- a/backend/tests/test_purge_cascade.py +++ b/backend/tests/test_purge_cascade.py @@ -5,6 +5,7 @@ import pytest from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError from app import settings as settings_mod from app.db.models import ( @@ -180,6 +181,22 @@ def test_tombstoned_alert_and_risk_are_hidden_from_read_apis(db_session, tmp_pat assert exc.value.status_code == 404 +def test_database_failure_prevents_unlink(db_session, tmp_path): + s = db_session + event_id, _, _, path = _mk_chain( + s, tmp_path, suffix="sql-failure", severity="high", age_days=0, + ) + s.add(Event( + event_id="invalid-extra", device_id="missing-device", source_type="image", + timestamp=datetime.now(UTC), + )) + with pytest.raises(IntegrityError): + purge.delete_events(s, [event_id]) + s.rollback() + assert path.exists() + assert s.get(Event, event_id).deletion_state == "active" + + def test_wipe_old_events_leaves_no_orphans(db_session, tmp_path): """The storage-wipe path (older_than_days) must clear the full chain.""" s = db_session From 22ca5f8b3aa0796a3e4f4ff4a57f0f3c497c5929 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 09:59:33 -0400 Subject: [PATCH 06/29] fix: enforce parent processing choices at worker boundaries --- agent-windows/src/broker_service.py | 55 +++++++++- agent-windows/src/main.py | 3 + agent-windows/tests/test_broker_service.py | 21 ++++ backend/app/api/consent.py | 19 +++- backend/app/api/devices.py | 8 +- backend/app/api/events.py | 36 +++--- backend/app/services/consent.py | 52 +++++++-- backend/app/services/daily_digest.py | 4 +- backend/app/services/guardian_review.py | 76 +++++++++++-- backend/app/services/notifications.py | 13 ++- backend/app/services/screenshot_async.py | 14 +++ backend/tests/test_processing_policy.py | 121 +++++++++++++++++++++ backend/tests/test_readiness_apis.py | 11 ++ 13 files changed, 390 insertions(+), 43 deletions(-) create mode 100644 backend/tests/test_processing_policy.py diff --git a/agent-windows/src/broker_service.py b/agent-windows/src/broker_service.py index a91967b..0459cd0 100644 --- a/agent-windows/src/broker_service.py +++ b/agent-windows/src/broker_service.py @@ -17,6 +17,7 @@ import threading import time from dataclasses import dataclass +from datetime import UTC, datetime from pathlib import Path from typing import Any, Protocol @@ -114,7 +115,31 @@ def __init__( self.replay_cache = replay_cache or RequestReplayCache() self._capture_config_lock = threading.Lock() self._capture_config: dict[str, Any] = {} - self.set_capture_config(capture_config or {}) + self.capture_config_path = self.pause_path.with_name("capture_config.json") + try: + cached = json.loads(self.capture_config_path.read_text(encoding="utf-8")) + if isinstance(cached, dict): + self.set_capture_config(cached) + except (OSError, ValueError): + pass + if capture_config: + self.set_capture_config(capture_config) + + def _server_paused(self) -> bool: + with self._capture_config_lock: + cfg = dict(self._capture_config) + if cfg.get("paused") is not True: + return False + until = cfg.get("paused_until") + if not until: + return True + try: + expiry = datetime.fromisoformat(until) + if expiry.tzinfo is None: + return True + return expiry.astimezone(UTC) > datetime.now(UTC) + except ValueError: + return True def handle_message(self, message: dict[str, Any]) -> dict[str, Any]: try: @@ -146,13 +171,15 @@ def handle_request(self, request: BrokerRequest) -> dict[str, Any]: "device_id": creds.get("device_id") or "", "backend_url": creds.get("backend_url") or "", "queue_depth": self.queue.qsize(), - "paused": pause.paused, + "paused": pause.paused or self._server_paused(), "paused_until": pause.paused_until, } if request.action == "capture_config": with self._capture_config_lock: return dict(self._capture_config) if request.action == "submit_screenshot": + if self._load_pause().paused or self._server_paused(): + raise ProtocolError("capture is paused by parent policy") payload = self._screenshot_payload(request.payload) self.queue.put_nowait(payload) return {"status": "queued", "queue_depth": self.queue.qsize()} @@ -177,8 +204,32 @@ def set_capture_config(self, payload: dict[str, Any]) -> None: full_screen = payload.get("full_screen_capture_enabled") if isinstance(full_screen, bool): clean["full_screen_capture_enabled"] = full_screen + paused = payload.get("paused") + if isinstance(paused, bool): + clean["paused"] = paused + capture_allowed = payload.get("capture_allowed") + if isinstance(capture_allowed, bool): + clean["capture_allowed"] = capture_allowed + if not capture_allowed: + clean["paused"] = True + context_allowed = payload.get("apps_and_urls_allowed") + if isinstance(context_allowed, bool): + clean["apps_and_urls_allowed"] = context_allowed + paused_until = payload.get("paused_until") + if isinstance(paused_until, str) and len(paused_until) <= 40: + clean["paused_until"] = paused_until + generation = payload.get("policy_generation") + if isinstance(generation, str) and len(generation) <= 64: + clean["policy_generation"] = generation with self._capture_config_lock: self._capture_config = clean + try: + self.capture_config_path.parent.mkdir(parents=True, exist_ok=True) + temporary = self.capture_config_path.with_suffix(".tmp") + temporary.write_text(json.dumps(clean), encoding="utf-8") + os.replace(temporary, self.capture_config_path) + except OSError as exc: + log.warning("capture policy could not be persisted: %s", type(exc).__name__) def ensure_broker_credentials(self) -> dict[str, Any]: creds = load_credentials(self.credential_path) diff --git a/agent-windows/src/main.py b/agent-windows/src/main.py index 13f9d24..9311ac2 100644 --- a/agent-windows/src/main.py +++ b/agent-windows/src/main.py @@ -140,6 +140,9 @@ async def capture_loop(cfg: AgentConfig, screenshot_queue: asyncio.Queue) -> Non ) if send: + if _is_locally_paused(): + await asyncio.sleep(cfg.ocr_cadence_seconds) + continue last_phash = shot.phash last_full_phash = shot.full_phash last_sent_at = time.time() diff --git a/agent-windows/tests/test_broker_service.py b/agent-windows/tests/test_broker_service.py index c41b05a..30bef43 100644 --- a/agent-windows/tests/test_broker_service.py +++ b/agent-windows/tests/test_broker_service.py @@ -3,6 +3,7 @@ import asyncio import inspect import json +from datetime import UTC, datetime, timedelta from types import SimpleNamespace from src.broker_protocol import image_to_b64, make_request @@ -70,6 +71,26 @@ def test_broker_returns_only_bounded_capture_config(tmp_path) -> None: assert "must-not-cross-the-pipe" not in json.dumps(response) +def test_parent_pause_reaches_capture_and_blocks_submit(tmp_path) -> None: + handler = _handler(tmp_path) + handler.set_capture_config({ + "level": "balanced", "cadence_seconds": 5, "paused": True, + "paused_until": (datetime.now(UTC) + timedelta(minutes=5)).isoformat(), + "capture_allowed": False, + "policy_generation": "generation-2", "device_token": "must-not-cross-the-pipe", + }) + config = handler.handle_message(make_request("capture_config"))["payload"] + assert config["paused"] is True + assert config["capture_allowed"] is False + assert config["policy_generation"] == "generation-2" + assert "device_token" not in config + assert handler.handle_message(make_request("status"))["payload"]["paused"] is True + restarted = _handler(tmp_path) + assert restarted.handle_message(make_request("status"))["payload"]["paused"] is True + restarted.set_capture_config({"paused": False, "capture_allowed": True}) + assert restarted.handle_message(make_request("status"))["payload"]["paused"] is False + + def test_broker_migrates_legacy_credentials_to_secure_path(tmp_path) -> None: handler = _handler(tmp_path) save_credentials("dev-1", "secret-token", "http://127.0.0.1:8787", handler.legacy_credential_path) diff --git a/backend/app/api/consent.py b/backend/app/api/consent.py index b33823c..378064a 100644 --- a/backend/app/api/consent.py +++ b/backend/app/api/consent.py @@ -9,7 +9,16 @@ from ulid import ULID from app.api.deps import current_user, get_db_dep, require_recent_auth -from app.db.models import ChildProfile, ConsentRecord, Device, Event, Setting, User +from app.db.models import ( + ChildProfile, + ConsentRecord, + Device, + Event, + GuardianReview, + NotificationJob, + Setting, + User, +) from app.services import consent, purge from app.services.audit import log_action @@ -113,6 +122,14 @@ def withdraw_consent( supersedes_id=previous.consent_id if previous else None, ) db.add(row) + db.query(GuardianReview).filter(GuardianReview.status.in_(["queued", "running"])).update( + {"status": "cancelled", "error_code": "consent_withdrawn", "completed_at": datetime.now(UTC)}, + synchronize_session=False, + ) + db.query(NotificationJob).filter(NotificationJob.status.in_(["queued", "retry"])).update( + {"status": "dead", "last_error": "consent withdrawn", "next_attempt_at": None}, + synchronize_session=False, + ) for device in db.query(Device).all(): device.paired = False device.token_hash = None diff --git a/backend/app/api/devices.py b/backend/app/api/devices.py index fd0bf15..31ec4c3 100644 --- a/backend/app/api/devices.py +++ b/backend/app/api/devices.py @@ -335,13 +335,17 @@ def capture_config( child's policy. The agent polls this so a parent can tighten or loosen monitoring from the dashboard without touching the kid's PC.""" from app.db.models import ChildProfile - from app.services import profile_policy + from app.services import consent, profile_policy prof = db.get(ChildProfile, device.profile_id) if device.profile_id else None policy = (prof.alert_policy or {}) if prof is not None else {} age = prof.age_group if prof is not None else "10_13" cfg = profile_policy.capture_settings(policy, age) - cfg["paused"] = is_device_paused(device) + processing = consent.effective_policy(db) + cfg["paused"] = is_device_paused(device) or not processing.screenshots cfg["paused_until"] = device.paused_until.isoformat() if device.paused_until else None + cfg["capture_allowed"] = processing.screenshots + cfg["apps_and_urls_allowed"] = processing.apps_and_urls + cfg["policy_generation"] = processing.generation db.commit() # persist any expired-pause cleanup done by is_device_paused return cfg diff --git a/backend/app/api/events.py b/backend/app/api/events.py index 637f375..b174bb0 100644 --- a/backend/app/api/events.py +++ b/backend/app/api/events.py @@ -13,7 +13,7 @@ from app.api.deps import current_device, current_user, get_db_dep from app.db.models import Device, Event, ScreenshotUploadReceipt, User -from app.services import encryption, event_ingest, screenshot_async +from app.services import consent, encryption, event_ingest, screenshot_async from app.services.device_state import is_device_paused from app.services.evidence_paths import UnsafeEvidencePathError, resolve_stored_evidence_path from app.services.input_bounds import InputBoundsError, sanitize_metadata @@ -104,7 +104,8 @@ async def ingest( db: Session = Depends(get_db_dep), device: Device = Depends(current_device), ): - if is_device_paused(device): + policy = consent.effective_policy(db) + if is_device_paused(device) or not policy.active: # Quietly accept but don't store — pause behavior is honored server-side too return IngestResponse( event_id=req.event_id or "", @@ -114,10 +115,13 @@ async def ingest( score=0, categories=[], ) + payload = req.model_dump() + if not policy.apps_and_urls: + payload.update(app_name=None, window_title=None, url=None) try: result = await event_ingest.ingest_event( db, - payload=req.model_dump(), + payload=payload, device_id=device.device_id, source_ip=request.client.host if request.client else None, ) @@ -167,7 +171,7 @@ def list_events( db: Session = Depends(get_db_dep), _: User = Depends(current_user), ): - q = db.query(Event).order_by(Event.timestamp.desc()) + q = db.query(Event).filter(Event.deletion_state == "active").order_by(Event.timestamp.desc()) if device_id: q = q.filter(Event.device_id == device_id) if profile_id: @@ -183,7 +187,7 @@ def get_event_text( user: User = Depends(current_user), ): e = db.get(Event, event_id) - if e is None: + if e is None or e.deletion_state != "active": raise HTTPException(404, "Not found") if not e.redacted_text_enc: return {"text": ""} @@ -262,7 +266,8 @@ async def ingest_screenshot( if existing_receipt is not None: return _receipt_response(existing_receipt) - if is_device_paused(device): + policy = consent.effective_policy(db) + if is_device_paused(device) or not policy.screenshots: # Honor pause server-side: drop quietly, don't store. return ScreenshotIngestResponse(event_id="", status="paused", queued=False) @@ -292,9 +297,9 @@ async def ingest_screenshot( image_bytes, { "device_id": device.device_id, - "app_name": app_name, - "window_title": window_title, - "url": url, + "app_name": app_name if policy.apps_and_urls else None, + "window_title": window_title if policy.apps_and_urls else None, + "url": url if policy.apps_and_urls else None, "profile_id": profile_id, "age_group": age_group, "capture_scope": capture_scope, @@ -305,6 +310,7 @@ async def ingest_screenshot( "idempotency_key": idempotency_key, "mime_type": mime_type, "source_ip": request.client.host if request.client else None, + "policy_generation": policy.generation, }, ) if not screenshot_async.enqueue_nowait(token): @@ -349,18 +355,18 @@ def get_event_screenshot( from app.db.models import EvidenceBlob e = db.get(Event, event_id) - if e is None or not e.screenshot_blob_id: + if e is None or e.deletion_state != "active" or not e.screenshot_blob_id: raise HTTPException(404, "No screenshot") blob = db.get(EvidenceBlob, e.screenshot_blob_id) - if blob is None: + if blob is None or blob.deletion_state != "active": raise HTTPException(404, "Blob missing") try: path = resolve_stored_evidence_path(blob.encrypted_path) pt = encryption.decrypt_blob_from_disk(path, aad=blob.blob_id.encode("ascii")) - except (FileNotFoundError, UnsafeEvidencePathError) as e: - raise HTTPException(404, "Blob missing") from e - except Exception as e: - raise HTTPException(500, "Decryption failed") from e + except (FileNotFoundError, UnsafeEvidencePathError) as exc: + raise HTTPException(404, "Blob missing") from exc + except Exception as exc: + raise HTTPException(500, "Decryption failed") from exc from app.services.audit import log_action log_action(db, actor=str(user.id), action="evidence.view", target=event_id) db.commit() diff --git a/backend/app/services/consent.py b/backend/app/services/consent.py index d798764..958a89a 100644 --- a/backend/app/services/consent.py +++ b/backend/app/services/consent.py @@ -1,6 +1,8 @@ """Append-only parental consent and monitoring gate.""" from __future__ import annotations +from dataclasses import dataclass + from sqlalchemy.orm import Session from app.db.models import ConsentRecord, User @@ -8,19 +10,47 @@ NOTICE_VERSION = "privacy-notice-2026-08-v1" -def latest(session: Session) -> ConsentRecord | None: - return session.query(ConsentRecord).order_by(ConsentRecord.created_at.desc()).first() +@dataclass(frozen=True) +class ProcessingPolicy: + generation: str | None + active: bool + screenshots: bool + apps_and_urls: bool + retention: bool + notifications: bool + external_ai: bool -def monitoring_allowed(session: Session) -> bool: - # Setup/bootstrap may run before the first administrator exists. Once the - # family account exists, explicit current-version consent is mandatory. +def effective_policy(session: Session) -> ProcessingPolicy: + """Resolve the latest parent choices at a processing boundary. + + Capture needs durable retention because its upload/classification pipeline + persists encrypted frames before analysis. Until an ephemeral pipeline is + implemented, disabling retention also disables new capture. + """ if session.query(User).count() == 0: - return True + return ProcessingPolicy(None, True, True, True, True, True, False) row = latest(session) - return bool( - row - and row.status == "granted" - and row.notice_version == NOTICE_VERSION - and row.choices.get("child_notice_acknowledged") is True + choices = row.choices or {} if row else {} + active = bool( + row and row.status == "granted" and row.notice_version == NOTICE_VERSION + and choices.get("child_notice_acknowledged") is True + ) + return ProcessingPolicy( + row.consent_id if row else None, + active, + active and choices.get("screenshots") is True and choices.get("retention") is True, + active and choices.get("apps_and_urls") is True, + active and choices.get("retention") is True, + active and choices.get("notifications") is True, + active and choices.get("external_ai") is True, ) + + +def latest(session: Session) -> ConsentRecord | None: + return session.query(ConsentRecord).order_by(ConsentRecord.created_at.desc()).first() + + +def monitoring_allowed(session: Session) -> bool: + policy = effective_policy(session) + return policy.active and policy.screenshots diff --git a/backend/app/services/daily_digest.py b/backend/app/services/daily_digest.py index bd410d4..5543255 100644 --- a/backend/app/services/daily_digest.py +++ b/backend/app/services/daily_digest.py @@ -10,7 +10,7 @@ from ulid import ULID from app.db.models import Alert, DigestRun, Event, RiskResult, Setting -from app.services import notifications +from app.services import consent, notifications def family_timezone(session: Session) -> str: @@ -100,6 +100,8 @@ def _body(summary: dict) -> str: def run_due(session: Session, *, now: datetime | None = None) -> DigestRun | None: + if not consent.effective_policy(session).notifications: + return None config = _notification_config(session) if not config.get("enabled") or not config.get("daily_digest_enabled", True): return None diff --git a/backend/app/services/guardian_review.py b/backend/app/services/guardian_review.py index f29bb83..73eef81 100644 --- a/backend/app/services/guardian_review.py +++ b/backend/app/services/guardian_review.py @@ -34,7 +34,7 @@ ReviewStatus, ReviewSummary, ) -from app.services import encryption +from app.services import consent, encryption from app.services import guardian_review_codex_auth as codex_auth from app.services.audit import log_action from app.services.guardian_review_minimization import InvalidIncidentError, build_minimized_incident @@ -136,6 +136,20 @@ def _ensure_provider_ready(provider: str) -> None: raise WorkflowError(str(readiness["blocking_reason"]), status_code=503) +def _ensure_review_policy(db: Session, provider: str, generation: str | None = None) -> None: + """Check the latest parent decision before each outbound review boundary.""" + policy = consent.effective_policy(db) + if generation is not None and policy.generation != generation: + raise WorkflowError("consent_changed", status_code=409) + # Mock mode is a local synthetic demonstration, but a recorded withdrawal + # still stops its queued work. Live providers require an explicit opt in. + current = consent.latest(db) + if current is not None and not policy.active: + raise WorkflowError("consent_required", status_code=409) + if provider != "mock" and not policy.external_ai: + raise WorkflowError("external_ai_not_allowed", status_code=409) + + def _audit_details( *, alert_id: str, @@ -173,6 +187,7 @@ def create_preview( if provider not in RETENTION_NOTICES: raise WorkflowError("configuration_error", status_code=503) _ensure_provider_ready(provider) + _ensure_review_policy(db, provider) schema_version = SCHEMA_VERSION prompt_version = _settings().guardian_review_prompt_version or PROMPT_VERSION try: @@ -193,6 +208,7 @@ def create_preview( created_at = _now() expires_at = created_at + timedelta(seconds=_settings().guardian_review_preview_ttl_seconds) encrypted_record = { + "policy_generation": consent.effective_policy(db).generation, "context": context.model_dump(mode="json"), "outbound_payload": incident.payload, "redactions": incident.redactions, @@ -287,6 +303,8 @@ def submit_review( raise WorkflowError("not_found", status_code=404) if preview.provider != _settings().guardian_review_provider.strip().lower(): raise WorkflowError("preview_stale", status_code=409) + stored = _load_preview_payload(preview) + _ensure_review_policy(db, preview.provider, stored.get("policy_generation")) _ensure_provider_ready(preview.provider) if preview.review_id: existing = db.get(GuardianReview, preview.review_id) @@ -296,7 +314,6 @@ def submit_review( raise WorkflowError("preview_expired", status_code=409) if not hmac.compare_digest(preview.payload_digest, preview_digest): raise WorkflowError("preview_stale", status_code=409) - stored = _load_preview_payload(preview) try: context = GuardianReviewContext.model_validate(stored["context"]) incident = build_minimized_incident( @@ -645,10 +662,16 @@ async def process_one(db: Session, *, provider_override=None, sleep=asyncio.slee row = db.query(GuardianReview).filter(GuardianReview.status == "queued").order_by(GuardianReview.created_at).first() if row is None: return False - row.status = "running" - row.started_at = _now() - row.error_code = None + claimed = db.query(GuardianReview).filter( + GuardianReview.review_id == row.review_id, GuardianReview.status == "queued" + ).update({"status": "running", "started_at": _now(), "error_code": None}, synchronize_session=False) db.commit() + if not claimed: + return False + db.expire_all() + row = db.get(GuardianReview, row.review_id) + assert row is not None + review_id = row.review_id preview = db.get(GuardianReviewPreview, row.preview_id) if preview is None: _fail(db, row, "preview_stale") @@ -662,6 +685,11 @@ async def process_one(db: Session, *, provider_override=None, sleep=asyncio.slee if not isinstance(payload, dict): _fail(db, row, "preview_stale") return True + try: + _ensure_review_policy(db, row.provider, stored.get("policy_generation")) + except WorkflowError as exc: + _fail(db, row, exc.code) + return True try: provider = provider_override or provider_for(_settings(), provider_name=row.provider) except ProviderError as exc: @@ -679,11 +707,29 @@ async def process_one(db: Session, *, provider_override=None, sleep=asyncio.slee status="sent", ) for attempt in range(1, max_attempts + 1): + db.expire_all() + row = db.get(GuardianReview, review_id) + if row is None or row.status != "running": + return True + try: + _ensure_review_policy(db, row.provider, stored.get("policy_generation")) + except WorkflowError as exc: + _fail(db, row, exc.code) + return True row.attempts = attempt log_action(db, actor="system", action="guardian_review.sent", target=row.review_id, details={**audit_base, "attempt": attempt}) db.commit() try: result = await provider.assess(payload=payload, prompt=_prompt(row.prompt_version), model=row.model_requested, timeout=float(_settings().guardian_review_timeout_seconds)) + db.expire_all() + row = db.get(GuardianReview, review_id) + if row is None or row.status != "running": + return True + try: + _ensure_review_policy(db, row.provider, stored.get("policy_generation")) + except WorkflowError as exc: + _fail(db, row, exc.code) + return True row.assessment_enc = encryption.encrypt_bytes(result.assessment.model_dump_json().encode(), aad=_review_aad(row.review_id)) row.model_returned = result.model_returned row.provider_response_id = result.response_id @@ -710,7 +756,14 @@ async def process_one(db: Session, *, provider_override=None, sleep=asyncio.slee return True -def _fail(db: Session, row: GuardianReview, code: str) -> None: +def _fail(db: Session, row: GuardianReview | None, code: str) -> None: + if row is None: + return + db.expire_all() + current = db.get(GuardianReview, row.review_id) + if current is None or current.status != "running": + return + row = current row.status = "failed" row.error_code = code row.completed_at = _now() @@ -744,8 +797,15 @@ def recover_stale_jobs(db: Session, *, older_than_seconds: int = 120) -> int: cutoff = _now() - timedelta(seconds=older_than_seconds) rows = db.query(GuardianReview).filter(GuardianReview.status == "running", GuardianReview.started_at < cutoff).all() for row in rows: - row.status = "queued" - row.started_at = None + if row.attempts: + # Once an attempt was recorded the external outcome is unknown. + # Retrying automatically could dispatch the same private context twice. + row.status = "failed" + row.error_code = "outcome_uncertain" + row.completed_at = _now() + else: + row.status = "queued" + row.started_at = None if rows: db.commit() return len(rows) diff --git a/backend/app/services/notifications.py b/backend/app/services/notifications.py index 6d9ecbc..d345704 100644 --- a/backend/app/services/notifications.py +++ b/backend/app/services/notifications.py @@ -24,7 +24,7 @@ from sqlalchemy.orm import Session from app.db.models import Alert, NotificationJob, NotificationLog, Setting -from app.services import encryption +from app.services import consent, encryption log = logging.getLogger(__name__) @@ -317,7 +317,7 @@ def dispatch( logs.append(nl) # 2) Email if SMTP configured and severity warrants immediate - cfg = _get_smtp_config(session) + cfg = _get_smtp_config(session) if consent.effective_policy(session).notifications else None if cfg and is_immediate: subject = f"[GuardianNode] {alert.severity.upper()} alert" body = ( @@ -384,7 +384,7 @@ def enqueue( detail=None, ) ) - if not is_immediate: + if not is_immediate or not consent.effective_policy(session).notifications: return None job = NotificationJob( alert_id=alert.alert_id, @@ -422,6 +422,13 @@ def process_pending(session: Session, *, limit: int = 10, now: datetime | None = ) processed = 0 for job in candidates: + if not consent.effective_policy(session).notifications: + job.status = "dead" + job.last_error = "notifications disallowed by current consent" + job.next_attempt_at = None + job.updated_at = now + processed += 1 + continue due = _as_aware_utc(job.next_attempt_at) if due is not None and due > now: continue diff --git a/backend/app/services/screenshot_async.py b/backend/app/services/screenshot_async.py index c0500b0..ccfb8c9 100644 --- a/backend/app/services/screenshot_async.py +++ b/backend/app/services/screenshot_async.py @@ -394,7 +394,10 @@ def _mtime(token: str) -> float: async def _classify_token(token: str) -> None: + from app.db.models import Device from app.db.session import get_sessionmaker + from app.services import consent + from app.services.device_state import is_device_paused meta = _load_meta(token) if meta is None: @@ -417,6 +420,17 @@ async def _classify_token(token: str) -> None: session = get_sessionmaker()() try: + policy = consent.effective_policy(session) + device = session.get(Device, meta["device_id"]) + if ( + not policy.screenshots + or policy.generation != meta.get("policy_generation") + or device is None + or not device.paired + or is_device_paused(device) + ): + _discard(token) + return await screenshot_ingest.ingest_screenshot( session, image_bytes=image_bytes, diff --git a/backend/tests/test_processing_policy.py b/backend/tests/test_processing_policy.py new file mode 100644 index 0000000..c886749 --- /dev/null +++ b/backend/tests/test_processing_policy.py @@ -0,0 +1,121 @@ +"""Parent choices must hold at enqueue and worker side-effect boundaries.""" +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime, timedelta + +import pytest + +from app import settings as settings_mod +from app.db.models import Alert, ConsentRecord, Device, Event, RiskResult, User +from app.guardian_review_models import GuardianReviewContext +from app.services import consent, encryption, guardian_review +from app.services.guardian_review_providers import ProviderError + + +def _incident(db): + user = User(display_name="Synthetic Parent", password_hash="synthetic", recovery_hash="synthetic") + db.add(user) + db.add(Device(device_id="synthetic-device", hostname="synthetic", paired=True)) + db.flush() + db.add(Event( + event_id="synthetic-event", device_id="synthetic-device", source_type="image", + timestamp=datetime.now(UTC), + redacted_text_enc=encryption.encrypt_text("Synthetic secrecy request"), + )) + db.add(RiskResult( + risk_id="synthetic-risk", event_id="synthetic-event", risk_level="high", + score=70, summary="Synthetic secrecy request", evidence=["Synthetic secrecy request"], + categories=["secrecy_request"], + )) + db.flush() + db.add(Alert( + alert_id="synthetic-alert", risk_id="synthetic-risk", device_id="synthetic-device", + severity="high", + )) + db.commit() + return user + + +def _consent(db, user, *, consent_id="generation-1", status="granted", **changes): + choices = { + "screenshots": True, "apps_and_urls": True, "retention": True, + "notifications": True, "external_ai": False, + "child_notice_acknowledged": True, + } + choices.update(changes) + db.add(ConsentRecord( + consent_id=consent_id, user_id=user.id, notice_version=consent.NOTICE_VERSION, + status=status, choices=choices, + created_at=datetime.now(UTC) + (timedelta(seconds=1) if status == "withdrawn" else timedelta()), + )) + db.commit() + + +def test_all_capture_choices_off_disable_monitoring(db_session): + user = _incident(db_session) + _consent(db_session, user, screenshots=False, apps_and_urls=False, retention=False, + notifications=False, external_ai=False) + assert consent.monitoring_allowed(db_session) is False + assert consent.effective_policy(db_session).apps_and_urls is False + + +def test_external_review_requires_global_opt_in(db_session, monkeypatch): + user = _incident(db_session) + _consent(db_session, user) + monkeypatch.setattr(settings_mod.settings, "guardian_review_enabled", True) + monkeypatch.setattr(settings_mod.settings, "guardian_review_provider", "openai") + monkeypatch.setattr(settings_mod.settings, "guardian_review_zdr_confirmed", True) + monkeypatch.setattr(settings_mod.settings, "openai_api_key", "synthetic-test-key") + with pytest.raises(guardian_review.WorkflowError) as exc: + guardian_review.create_preview( + db_session, alert_id="synthetic-alert", context=GuardianReviewContext(), user=user, + ) + assert exc.value.code == "external_ai_not_allowed" + + +def test_withdrawal_fences_queued_review_before_provider(db_session, monkeypatch): + user = _incident(db_session) + _consent(db_session, user, external_ai=True) + monkeypatch.setattr(settings_mod.settings, "guardian_review_enabled", True) + monkeypatch.setattr(settings_mod.settings, "guardian_review_provider", "mock") + preview = guardian_review.create_preview( + db_session, alert_id="synthetic-alert", context=GuardianReviewContext(), user=user, + ) + guardian_review.submit_review( + db_session, alert_id="synthetic-alert", preview_id=preview.preview_id, + preview_digest=preview.preview_digest, user=user, + ) + _consent(db_session, user, consent_id="withdrawn", status="withdrawn") + calls = [] + + class Spy: + async def assess(self, **kwargs): + calls.append(kwargs) + raise ProviderError("upstream_unavailable") + + asyncio.run(guardian_review.process_one(db_session, provider_override=Spy())) + assert calls == [] + + +def test_stale_running_review_with_dispatched_attempt_is_not_retried(db_session): + from app.db.models import GuardianReview, GuardianReviewPreview + + user = _incident(db_session) + db_session.add(GuardianReviewPreview( + preview_id="p", alert_id="synthetic-alert", actor_user_id=user.id, + provider="mock", model_requested="mock", schema_version="1.1.0", + prompt_version="guardian-review-v2", payload_digest="0" * 64, + incident_fingerprint="1" * 64, expires_at=datetime.now(UTC) + timedelta(minutes=10), + )) + db_session.flush() + db_session.add(GuardianReview( + review_id="r", preview_id="p", alert_id="synthetic-alert", + requester_user_id=user.id, provider="mock", dedup_key="2" * 64, + schema_version="1.1.0", prompt_version="guardian-review-v2", + model_requested="mock", status="running", attempts=1, + started_at=datetime.now(UTC) - timedelta(minutes=5), + )) + db_session.commit() + assert guardian_review.recover_stale_jobs(db_session) == 1 + assert db_session.get(GuardianReview, "r").error_code == "outcome_uncertain" diff --git a/backend/tests/test_readiness_apis.py b/backend/tests/test_readiness_apis.py index 51f2cc3..a5409dc 100644 --- a/backend/tests/test_readiness_apis.py +++ b/backend/tests/test_readiness_apis.py @@ -249,6 +249,17 @@ def test_alert_actions_are_effective_or_explicitly_unsupported(monkeypatch, tmp_ client = _client(monkeypatch, tmp_path) from app.db.session import get_sessionmaker + notice_version = client.get("/api/consent").json()["notice_version"] + consent_response = client.post( + "/api/consent", + json={"notice_version": notice_version, "choices": { + "screenshots": True, "apps_and_urls": True, "retention": True, + "notifications": True, "external_ai": False, + "child_notice_acknowledged": True, + }}, + ) + assert consent_response.status_code == 200 + db = get_sessionmaker()() try: db.add(Device(device_id="action-dev", hostname="kid-pc", paired=True)) From 89e81954d2ea5f7884db4811c9fbd7377b80b9ae Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:00:49 -0400 Subject: [PATCH 07/29] fix: require open child requests for parent decisions --- backend/app/api/child_requests.py | 29 +++++++++++--- .../tests/test_child_request_transitions.py | 40 +++++++++++++++++++ dashboard/src/pages/Requests.tsx | 2 +- 3 files changed, 65 insertions(+), 6 deletions(-) create mode 100644 backend/tests/test_child_request_transitions.py diff --git a/backend/app/api/child_requests.py b/backend/app/api/child_requests.py index 8777404..1702163 100644 --- a/backend/app/api/child_requests.py +++ b/backend/app/api/child_requests.py @@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request from pydantic import BaseModel, ConfigDict, Field -from sqlalchemy import and_, or_ +from sqlalchemy import and_, or_, update from sqlalchemy.orm import Session from ulid import ULID @@ -167,10 +167,28 @@ def review_child_request( row = db.get(ChildRequest, request_id) if row is None: raise HTTPException(404, "Request not found") - row.status = req.status - row.response_note = req.response_note - row.reviewed_by = str(user.id) - row.reviewed_at = datetime.now(UTC) + now = datetime.now(UTC) + # A request can be answered only once while it is still open. The SQL + # transition also prevents two concurrent parent sessions overwriting one + # another's decision. + changed = db.execute( + update(ChildRequest) + .where( + ChildRequest.request_id == request_id, + ChildRequest.status == "open", + or_(ChildRequest.expires_at.is_(None), ChildRequest.expires_at > now), + ) + .execution_options(synchronize_session=False) + .values( + status=req.status, + response_note=req.response_note, + reviewed_by=str(user.id), + reviewed_at=now, + ) + ).rowcount + if changed != 1: + db.rollback() + raise HTTPException(409, "Request is closed or expired") log_action( db, actor=str(user.id), @@ -180,4 +198,5 @@ def review_child_request( source_ip=request.client.host if request.client else None, ) db.commit() + db.refresh(row) return row diff --git a/backend/tests/test_child_request_transitions.py b/backend/tests/test_child_request_transitions.py new file mode 100644 index 0000000..98f7ef2 --- /dev/null +++ b/backend/tests/test_child_request_transitions.py @@ -0,0 +1,40 @@ +from datetime import UTC, datetime, timedelta + +import pytest +from fastapi import HTTPException, Request + +from app.api.child_requests import ChildRequestReview, review_child_request +from app.db.models import ChildRequest, User + + +def _request() -> Request: + return Request({ + "type": "http", "method": "POST", "path": "/api/child-requests/r/review", + "headers": [], "client": ("127.0.0.1", 1234), "scheme": "http", + "server": ("testserver", 80), + }) + + +def test_parent_decision_cannot_overwrite_or_approve_expired_request(db_session): + user = User(display_name="Synthetic Parent", password_hash="test", recovery_hash="test") + db_session.add(user) + db_session.add(ChildRequest( + request_id="open", request_type="more_time", status="open", + expires_at=datetime.now(UTC) + timedelta(days=1), + )) + db_session.add(ChildRequest( + request_id="expired", request_type="more_time", status="open", + expires_at=datetime.now(UTC) - timedelta(seconds=1), + )) + db_session.commit() + + decided = review_child_request("open", ChildRequestReview(status="approved"), _request(), db_session, user) + assert decided.status == "approved" + + for request_id in ("open", "expired"): + with pytest.raises(HTTPException) as conflict: + review_child_request(request_id, ChildRequestReview(status="denied"), _request(), db_session, user) + assert conflict.value.status_code == 409 + db_session.expire_all() + assert db_session.get(ChildRequest, "open").status == "approved" + assert db_session.get(ChildRequest, "expired").status == "open" diff --git a/dashboard/src/pages/Requests.tsx b/dashboard/src/pages/Requests.tsx index a23dfa9..304f9ac 100644 --- a/dashboard/src/pages/Requests.tsx +++ b/dashboard/src/pages/Requests.tsx @@ -16,7 +16,7 @@ export default function Requests() { const rows = requests.data!.items; return (
-

Family conversation

Requests

Children can ask for more time or an exception. A parent always makes the decision.

+

Family conversation

Requests

Children can ask for more time or an exception. Your decision is recorded and shared with the child; it does not automatically change device rules.

All requests

{requests.data!.open_count} open
{rows.length ?
{rows.map((row) => review.mutate({ id: row.request_id, status })} />)}
:
} From 047752d6737afc0667b94caf915329e4483563a9 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:01:17 -0400 Subject: [PATCH 08/29] fix: deliver withdrawal stop through control-only device credential --- backend/app/api/consent.py | 11 +++-- backend/app/api/deps.py | 4 ++ backend/app/api/devices.py | 12 +++++- backend/tests/test_processing_policy.py | 56 ++++++++++++++++++++++++- 4 files changed, 78 insertions(+), 5 deletions(-) diff --git a/backend/app/api/consent.py b/backend/app/api/consent.py index 378064a..ccf8d6e 100644 --- a/backend/app/api/consent.py +++ b/backend/app/api/consent.py @@ -131,9 +131,14 @@ def withdraw_consent( synchronize_session=False, ) for device in db.query(Device).all(): - device.paired = False - device.token_hash = None - device.status = "consent_withdrawn" + # Keep the existing token for control polling until the broker confirms + # it applied this generation. Other device endpoints reject it. + if device.paired and device.token_hash: + device.status = "consent_stop_requested" + else: + device.paired = False + device.token_hash = None + device.status = "consent_stop_offline" deletion = None if req.evidence_disposition == "delete": deletion = purge.delete_events(db, [item[0] for item in db.query(Event.event_id).all()]) diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index 590f41b..64b1ae4 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -173,6 +173,10 @@ def current_device(request: Request, db: Session = Depends(get_db_dep)) -> Devic device = device_tokens.authenticate(db, token) if device is not None: rate_limit.reset("device_auth", client_ip) + if device.status == "consent_stop_requested" and request.url.path not in { + "/api/devices/capture-config", "/api/devices/heartbeat", + }: + raise HTTPException(status_code=403, detail="Consent withdrawn; control polling only") operation = device_budget.operation_for_path(request.url.path, request.method) if operation is not None: try: diff --git a/backend/app/api/devices.py b/backend/app/api/devices.py index 31ec4c3..1c3a539 100644 --- a/backend/app/api/devices.py +++ b/backend/app/api/devices.py @@ -353,6 +353,7 @@ def capture_config( class HeartbeatRequest(BaseModel): queued_frames: int = Field(default=0, ge=0, le=10000) agent_version: str | None = None + applied_policy_generation: str | None = Field(default=None, max_length=64) @router.post("/heartbeat") @@ -365,7 +366,16 @@ def heartbeat( from app.services import pipeline_metrics paused_until = effective_paused_until(device) # clears expired pauses device.last_seen = datetime.now(UTC) - if device.status not in ("paused", "disabled"): + if device.status == "consent_stop_requested": + latest = consent.latest(db) + if ( + latest is not None and latest.status == "withdrawn" + and req.applied_policy_generation == latest.consent_id + ): + device.paired = False + device.token_hash = None + device.status = "consent_stop_applied" + elif device.status not in ("paused", "disabled"): device.status = "online" if req.agent_version: device.agent_version = req.agent_version diff --git a/backend/tests/test_processing_policy.py b/backend/tests/test_processing_policy.py index c886749..1bb5c96 100644 --- a/backend/tests/test_processing_policy.py +++ b/backend/tests/test_processing_policy.py @@ -5,11 +5,15 @@ from datetime import UTC, datetime, timedelta import pytest +from fastapi import HTTPException, Request from app import settings as settings_mod +from app.api.consent import ConsentWithdrawalRequest, withdraw_consent +from app.api.deps import current_device +from app.api.devices import HeartbeatRequest, heartbeat from app.db.models import Alert, ConsentRecord, Device, Event, RiskResult, User from app.guardian_review_models import GuardianReviewContext -from app.services import consent, encryption, guardian_review +from app.services import consent, device_tokens, encryption, guardian_review from app.services.guardian_review_providers import ProviderError @@ -119,3 +123,53 @@ def test_stale_running_review_with_dispatched_attempt_is_not_retried(db_session) db_session.commit() assert guardian_review.recover_stale_jobs(db_session) == 1 assert db_session.get(GuardianReview, "r").error_code == "outcome_uncertain" + + +def _request(path: str, token: str = "") -> Request: + return Request({ + "type": "http", "method": "POST", "path": path, + "headers": [(b"authorization", f"Bearer {token}".encode())] if token else [], + "client": ("127.0.0.1", 1234), "scheme": "http", "server": ("testserver", 80), + }) + + +def test_withdrawal_token_is_control_only_until_broker_ack(db_session): + user = _incident(db_session) + _consent(db_session, user) + token, digest = device_tokens.issue_token("synthetic-device") + device = db_session.get(Device, "synthetic-device") + device.token_hash = digest + db_session.commit() + + result = withdraw_consent( + ConsentWithdrawalRequest(evidence_disposition="retain"), + _request("/api/consent/withdraw"), db_session, user, None, + ) + generation = result["record"]["consent_id"] + assert device.status == "consent_stop_requested" + assert device.paired is True + with pytest.raises(HTTPException) as denied: + current_device(_request("/api/events/screenshot", token), db_session) + assert denied.value.status_code == 403 + control_device = current_device(_request("/api/devices/capture-config", token), db_session) + assert control_device.device_id == device.device_id + + heartbeat(HeartbeatRequest(applied_policy_generation="stale-generation"), db_session, device) + assert device.status == "consent_stop_requested" + heartbeat(HeartbeatRequest(applied_policy_generation=generation), db_session, device) + assert device.status == "consent_stop_applied" + assert device.token_hash is None + with pytest.raises(HTTPException) as stale: + current_device(_request("/api/devices/capture-config", token), db_session) + assert stale.value.status_code == 401 + + +def test_withdrawal_marks_unpaired_device_offline(db_session): + user = _incident(db_session) + _consent(db_session, user) + result = withdraw_consent( + ConsentWithdrawalRequest(evidence_disposition="retain"), + _request("/api/consent/withdraw"), db_session, user, None, + ) + assert result["active"] is False + assert db_session.get(Device, "synthetic-device").status == "consent_stop_offline" From 42c9e2683016f5fa68b583febfa96b234a18bdb2 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:01:48 -0400 Subject: [PATCH 09/29] test: cover failed parent step-up in browser --- dashboard/e2e/family-dashboard.spec.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/dashboard/e2e/family-dashboard.spec.ts b/dashboard/e2e/family-dashboard.spec.ts index 656b189..ca95d6c 100644 --- a/dashboard/e2e/family-dashboard.spec.ts +++ b/dashboard/e2e/family-dashboard.spec.ts @@ -243,9 +243,13 @@ test("protected pause uses masked step-up and resumes only after confirmation", expect(pauseCalls).toBe(1); await page.getByRole("button", { name: "Pause", exact: true }).click(); + await dialog.getByLabel("Parent password").fill("incorrect"); + await dialog.getByRole("button", { name: "Continue" }).click(); + await expect(dialog.getByRole("alert")).toContainText("Invalid password"); + expect(pauseCalls).toBe(2); await dialog.getByLabel("Parent password").fill("synthetic-browser-password"); await dialog.getByRole("button", { name: "Continue" }).click(); await expect(page.getByText("Monitoring pause recorded.")).toBeVisible(); expect(pauseCalls).toBe(3); - expect(errors.filter((message) => !message.includes("status of 403 (Forbidden)"))).toEqual([]); + expect(errors.filter((message) => !/status of (401|403) \(/.test(message))).toEqual([]); }); From b0cf4db7c9c450a47be47818254e7ed10163c2c2 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:02:12 -0400 Subject: [PATCH 10/29] fix: require stop acknowledgment before regrant can resume device --- backend/app/api/devices.py | 23 ++++++++++++++++++++--- backend/tests/test_processing_policy.py | 12 ++++++++++-- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/backend/app/api/devices.py b/backend/app/api/devices.py index 1c3a539..8fc3881 100644 --- a/backend/app/api/devices.py +++ b/backend/app/api/devices.py @@ -12,7 +12,7 @@ from app import settings as settings_mod from app.api.deps import current_device, current_user, get_db_dep, require_recent_auth -from app.db.models import Device, DeviceCommand, PairingCode, User +from app.db.models import ConsentRecord, Device, DeviceCommand, PairingCode, User from app.db.session import begin_immediate_if_sqlite from app.services import consent, device_tokens, rate_limit from app.services import pairing as pairing_svc @@ -346,6 +346,18 @@ def capture_config( cfg["capture_allowed"] = processing.screenshots cfg["apps_and_urls_allowed"] = processing.apps_and_urls cfg["policy_generation"] = processing.generation + if device.status == "consent_stop_requested": + # A later re-grant does not silently resume a device that has not yet + # acknowledged the withdrawal. It must stop and then re-pair. + withdrawn = ( + db.query(ConsentRecord) + .filter(ConsentRecord.status == "withdrawn") + .order_by(ConsentRecord.created_at.desc()) + .first() + ) + cfg["paused"] = True + cfg["capture_allowed"] = False + cfg["policy_generation"] = withdrawn.consent_id if withdrawn else processing.generation db.commit() # persist any expired-pause cleanup done by is_device_paused return cfg @@ -367,9 +379,14 @@ def heartbeat( paused_until = effective_paused_until(device) # clears expired pauses device.last_seen = datetime.now(UTC) if device.status == "consent_stop_requested": - latest = consent.latest(db) + latest = ( + db.query(ConsentRecord) + .filter(ConsentRecord.status == "withdrawn") + .order_by(ConsentRecord.created_at.desc()) + .first() + ) if ( - latest is not None and latest.status == "withdrawn" + latest is not None and req.applied_policy_generation == latest.consent_id ): device.paired = False diff --git a/backend/tests/test_processing_policy.py b/backend/tests/test_processing_policy.py index 1bb5c96..d4cdaed 100644 --- a/backend/tests/test_processing_policy.py +++ b/backend/tests/test_processing_policy.py @@ -10,7 +10,7 @@ from app import settings as settings_mod from app.api.consent import ConsentWithdrawalRequest, withdraw_consent from app.api.deps import current_device -from app.api.devices import HeartbeatRequest, heartbeat +from app.api.devices import HeartbeatRequest, capture_config, heartbeat from app.db.models import Alert, ConsentRecord, Device, Event, RiskResult, User from app.guardian_review_models import GuardianReviewContext from app.services import consent, device_tokens, encryption, guardian_review @@ -51,7 +51,10 @@ def _consent(db, user, *, consent_id="generation-1", status="granted", **changes db.add(ConsentRecord( consent_id=consent_id, user_id=user.id, notice_version=consent.NOTICE_VERSION, status=status, choices=choices, - created_at=datetime.now(UTC) + (timedelta(seconds=1) if status == "withdrawn" else timedelta()), + created_at=datetime.now(UTC) + ( + timedelta(seconds=2) if consent_id == "later-grant" + else timedelta(seconds=1) if status == "withdrawn" else timedelta() + ), )) db.commit() @@ -156,6 +159,11 @@ def test_withdrawal_token_is_control_only_until_broker_ack(db_session): heartbeat(HeartbeatRequest(applied_policy_generation="stale-generation"), db_session, device) assert device.status == "consent_stop_requested" + _consent(db_session, user, consent_id="later-grant") + config = capture_config(db_session, device) + assert config["paused"] is True + assert config["capture_allowed"] is False + assert config["policy_generation"] == generation heartbeat(HeartbeatRequest(applied_policy_generation=generation), db_session, device) assert device.status == "consent_stop_applied" assert device.token_hash is None From ab5655f21763ec47becd3a554bb00fdaf40e35ac Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:02:56 -0400 Subject: [PATCH 11/29] fix: resolve capture and context choices independently --- backend/app/api/events.py | 2 +- backend/app/services/consent.py | 4 ++-- backend/tests/test_processing_policy.py | 17 +++++++++++++++++ 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/backend/app/api/events.py b/backend/app/api/events.py index b174bb0..1789d37 100644 --- a/backend/app/api/events.py +++ b/backend/app/api/events.py @@ -105,7 +105,7 @@ async def ingest( device: Device = Depends(current_device), ): policy = consent.effective_policy(db) - if is_device_paused(device) or not policy.active: + if is_device_paused(device) or not (policy.screenshots or policy.apps_and_urls): # Quietly accept but don't store — pause behavior is honored server-side too return IngestResponse( event_id=req.event_id or "", diff --git a/backend/app/services/consent.py b/backend/app/services/consent.py index 958a89a..36d6b21 100644 --- a/backend/app/services/consent.py +++ b/backend/app/services/consent.py @@ -40,7 +40,7 @@ def effective_policy(session: Session) -> ProcessingPolicy: row.consent_id if row else None, active, active and choices.get("screenshots") is True and choices.get("retention") is True, - active and choices.get("apps_and_urls") is True, + active and choices.get("apps_and_urls") is True and choices.get("retention") is True, active and choices.get("retention") is True, active and choices.get("notifications") is True, active and choices.get("external_ai") is True, @@ -53,4 +53,4 @@ def latest(session: Session) -> ConsentRecord | None: def monitoring_allowed(session: Session) -> bool: policy = effective_policy(session) - return policy.active and policy.screenshots + return policy.active and (policy.screenshots or policy.apps_and_urls) diff --git a/backend/tests/test_processing_policy.py b/backend/tests/test_processing_policy.py index d4cdaed..8a862af 100644 --- a/backend/tests/test_processing_policy.py +++ b/backend/tests/test_processing_policy.py @@ -67,6 +67,23 @@ def test_all_capture_choices_off_disable_monitoring(db_session): assert consent.effective_policy(db_session).apps_and_urls is False +def test_capture_and_context_choices_resolve_independently(db_session): + user = _incident(db_session) + _consent(db_session, user, screenshots=False, apps_and_urls=True) + policy = consent.effective_policy(db_session) + assert policy.screenshots is False + assert policy.apps_and_urls is True + assert consent.monitoring_allowed(db_session) is True + db_session.query(ConsentRecord).delete() + db_session.commit() + _consent(db_session, user, consent_id="no-retention", screenshots=True, + apps_and_urls=True, retention=False) + policy = consent.effective_policy(db_session) + assert policy.screenshots is False + assert policy.apps_and_urls is False + assert consent.monitoring_allowed(db_session) is False + + def test_external_review_requires_global_opt_in(db_session, monkeypatch): user = _incident(db_session) _consent(db_session, user) From 1dd5775a009d847e5764a87fc78b9b63a4c1e54e Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:03:22 -0400 Subject: [PATCH 12/29] fix: recover broker credentials and bind IPC responses --- agent-windows/src/backend_client.py | 7 ++- agent-windows/src/broker_client.py | 34 +++++++++++--- agent-windows/src/broker_service.py | 44 ++++++++++++++---- agent-windows/src/main.py | 9 ++-- agent-windows/tests/test_broker_client.py | 53 ++++++++++++++++++++++ agent-windows/tests/test_broker_service.py | 12 +++++ agent-windows/tests/test_sender_retry.py | 14 ++++++ 7 files changed, 153 insertions(+), 20 deletions(-) create mode 100644 agent-windows/tests/test_broker_client.py diff --git a/agent-windows/src/backend_client.py b/agent-windows/src/backend_client.py index 037da73..86d03a9 100644 --- a/agent-windows/src/backend_client.py +++ b/agent-windows/src/backend_client.py @@ -93,15 +93,18 @@ async def send_screenshot( r.raise_for_status() return r.json() - async def heartbeat(self, queued_frames: int = 0) -> bool: + async def heartbeat(self, queued_frames: int = 0, applied_policy_generation: str | None = None) -> bool: """Liveness + upload-backlog report. Falls back to the plain health probe when talking to a backend that predates /api/devices/heartbeat.""" try: async with httpx.AsyncClient(timeout=5.0, verify=self._verify()) as c: + body: dict[str, int | str] = {"queued_frames": queued_frames} + if applied_policy_generation: + body["applied_policy_generation"] = applied_policy_generation r = await c.post( f"{self.base_url}/api/devices/heartbeat", headers=self._headers(), - json={"queued_frames": queued_frames}, + json=body, ) if r.status_code in (404, 405): r = await c.get(f"{self.base_url}/api/health", headers=self._headers()) diff --git a/agent-windows/src/broker_client.py b/agent-windows/src/broker_client.py index 66cad4e..4f72cde 100644 --- a/agent-windows/src/broker_client.py +++ b/agent-windows/src/broker_client.py @@ -5,7 +5,9 @@ from typing import Any from src.broker_protocol import ( + MAX_MESSAGE_BYTES, PIPE_NAME, + PROTOCOL_VERSION, ProtocolError, decode_frame, encode_frame, @@ -29,7 +31,8 @@ def request(self, action: str, payload: dict[str, Any] | None = None) -> dict[st import win32file # type: ignore import win32pipe # type: ignore - frame = encode_frame(make_request(action, payload)) + request = make_request(action, payload) + frame = encode_frame(request) try: win32pipe.WaitNamedPipe(self.pipe_name, self.timeout_ms) handle = win32file.CreateFile( @@ -45,7 +48,7 @@ def request(self, action: str, payload: dict[str, Any] | None = None) -> dict[st raise BrokerUnavailable(str(exc)) from exc try: win32file.WriteFile(handle, frame) - response = self._read_response(handle) + response = self._read_response(handle, request["request_id"]) finally: win32file.CloseHandle(handle) if not response.get("ok"): @@ -56,13 +59,32 @@ def request(self, action: str, payload: dict[str, Any] | None = None) -> dict[st return payload_obj @staticmethod - def _read_response(handle) -> dict[str, Any]: # noqa: ANN001 + def _read_exact(handle, size: int) -> bytes: # noqa: ANN001 import win32file # type: ignore - _, header = win32file.ReadFile(handle, 4) + data = bytearray() + while len(data) < size: + _, chunk = win32file.ReadFile(handle, size - len(data)) + if not chunk: + raise ProtocolError("broker disconnected during response") + data.extend(chunk) + return bytes(data) + + @staticmethod + def _read_response(handle, request_id: str) -> dict[str, Any]: # noqa: ANN001 + header = BrokerClient._read_exact(handle, 4) size = int.from_bytes(header, "big") - _, body = win32file.ReadFile(handle, size) - return decode_frame(header + body) + if size > MAX_MESSAGE_BYTES: + raise ProtocolError("broker response too large") + body = BrokerClient._read_exact(handle, size) + response = decode_frame(header + body) + if response.get("version") != PROTOCOL_VERSION: + raise ProtocolError("unsupported broker protocol version") + if response.get("request_id") != request_id: + raise ProtocolError("broker response request_id mismatch") + if not isinstance(response.get("ok"), bool): + raise ProtocolError("broker returned invalid response") + return response def health(self) -> dict[str, Any]: return self.request("health") diff --git a/agent-windows/src/broker_service.py b/agent-windows/src/broker_service.py index 0459cd0..aff23f9 100644 --- a/agent-windows/src/broker_service.py +++ b/agent-windows/src/broker_service.py @@ -115,6 +115,7 @@ def __init__( self.replay_cache = replay_cache or RequestReplayCache() self._capture_config_lock = threading.Lock() self._capture_config: dict[str, Any] = {} + self._persisted_policy_generation: str | None = None self.capture_config_path = self.pause_path.with_name("capture_config.json") try: cached = json.loads(self.capture_config_path.read_text(encoding="utf-8")) @@ -223,14 +224,22 @@ def set_capture_config(self, payload: dict[str, Any]) -> None: clean["policy_generation"] = generation with self._capture_config_lock: self._capture_config = clean + self._persisted_policy_generation = None try: self.capture_config_path.parent.mkdir(parents=True, exist_ok=True) temporary = self.capture_config_path.with_suffix(".tmp") temporary.write_text(json.dumps(clean), encoding="utf-8") os.replace(temporary, self.capture_config_path) + with self._capture_config_lock: + if self._capture_config == clean: + self._persisted_policy_generation = clean.get("policy_generation") except OSError as exc: log.warning("capture policy could not be persisted: %s", type(exc).__name__) + def applied_policy_generation(self) -> str | None: + with self._capture_config_lock: + return self._persisted_policy_generation + def ensure_broker_credentials(self) -> dict[str, Any]: creds = load_credentials(self.credential_path) if creds and creds.get("device_token"): @@ -532,12 +541,28 @@ async def _run_sender(handler: BrokerCommandHandler, cfg: AgentConfig) -> None: token = str(creds.get("device_token") or "") backend_url = str(creds.get("backend_url") or cfg.backend_url) if not token: + try: + await asyncio.to_thread( + bootstrap_pairing, + socket.gethostname(), + __version__, + pending_path=pending_pairing_path(), + device_path=handler.credential_path, + bootstrap_token_path=broker_bootstrap_token_path(), + attempts=1, + ) + except Exception as exc: + log.warning("broker pairing retry failed: %s", type(exc).__name__) await asyncio.sleep(10) continue - await screenshot_sender_loop( - BackendClient(backend_url, token, ca_path=creds.get("ca_path")), - handler.queue, - ) # type: ignore[arg-type] + try: + await screenshot_sender_loop( + BackendClient(backend_url, token, ca_path=creds.get("ca_path")), + handler.queue, + ) # type: ignore[arg-type] + except Exception as exc: + log.warning("broker sender restarted after %s", type(exc).__name__) + await asyncio.sleep(10) async def _run_device_control(handler: BrokerCommandHandler, cfg: AgentConfig) -> None: @@ -558,7 +583,10 @@ async def _run_device_control(handler: BrokerCommandHandler, cfg: AgentConfig) - ca_path=creds.get("ca_path"), ) try: - await client.heartbeat(queued_frames=handler.queue.qsize()) + await client.heartbeat( + queued_frames=handler.queue.qsize(), + applied_policy_generation=handler.applied_policy_generation(), + ) capture_config = await client.get_capture_config() if capture_config: handler.set_capture_config(capture_config) @@ -637,6 +665,9 @@ def cli() -> None: force=True, ) cfg = AgentConfig.from_path(Path(args.config)) + if args.self_test: + log.info("broker self-test ok") + return handler = build_handler(cfg) try: bootstrap_pairing( @@ -649,9 +680,6 @@ def cli() -> None: except Exception as exc: log.warning("broker pairing bootstrap failed: %s", exc) handler.ensure_broker_credentials() - if args.self_test: - log.info("broker self-test ok") - return if os.name != "nt": raise SystemExit("GuardianNode endpoint broker service currently runs only on Windows") diff --git a/agent-windows/src/main.py b/agent-windows/src/main.py index 9311ac2..3ae7f95 100644 --- a/agent-windows/src/main.py +++ b/agent-windows/src/main.py @@ -240,8 +240,8 @@ async def screenshot_sender_loop(client: BackendClient, screenshot_queue: asynci except httpx.HTTPStatusError as e: status = e.response.status_code if status in (401, 403): - log.error("screenshot upload unauthorized (%s); re-pair this device", status) - backoff = 60.0 + log.error("screenshot upload unauthorized (%s); refreshing broker credentials", status) + backoff = 10.0 _retry_payload( screenshot_queue, payload, @@ -249,8 +249,9 @@ async def screenshot_sender_loop(client: BackendClient, screenshot_queue: asynci error=f"unauthorized: {status}", ) handled = True - await asyncio.sleep(backoff) - continue + # Return to the broker supervisor so a repaired pairing or + # rotated trust bundle creates a fresh BackendClient. + return if status == 413: log.error("screenshot rejected as too large; discarding frame") backoff = 1.0 diff --git a/agent-windows/tests/test_broker_client.py b/agent-windows/tests/test_broker_client.py new file mode 100644 index 0000000..270116d --- /dev/null +++ b/agent-windows/tests/test_broker_client.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import sys +from types import SimpleNamespace + +import pytest + +from src.broker_client import BrokerClient +from src.broker_protocol import MAX_MESSAGE_BYTES, ProtocolError, encode_frame, make_response + + +def _response_reader(monkeypatch, chunks: list[bytes]): + reads: list[int] = [] + + def read(_handle, size: int): + reads.append(size) + chunk = chunks.pop(0) + assert len(chunk) <= size + return 0, chunk + + monkeypatch.setitem(sys.modules, "win32file", SimpleNamespace(ReadFile=read)) + return reads + + +def test_response_reader_assembles_partial_header_and_body(monkeypatch): + frame = encode_frame(make_response("request-1", ok=True, payload={"status": "queued"})) + _response_reader(monkeypatch, [frame[:2], frame[2:4], frame[4:10], frame[10:]]) + assert BrokerClient._read_response(object(), "request-1")["payload"] == {"status": "queued"} + + +def test_response_reader_rejects_large_length_before_body_read(monkeypatch): + reads = _response_reader(monkeypatch, [(MAX_MESSAGE_BYTES + 1).to_bytes(4, "big")]) + with pytest.raises(ProtocolError, match="too large"): + BrokerClient._read_response(object(), "request-1") + assert reads == [4] + + +@pytest.mark.parametrize("frame", [ + encode_frame(make_response("other-request", ok=True)), + encode_frame({"version": 99, "request_id": "request-1", "ok": True, "payload": {}}), + encode_frame({"version": 1, "request_id": "request-1", "ok": "yes", "payload": {}}), +]) +def test_response_reader_rejects_unbound_or_invalid_response(monkeypatch, frame): + _response_reader(monkeypatch, [frame[:4], frame[4:]]) + with pytest.raises(ProtocolError): + BrokerClient._read_response(object(), "request-1") + + +def test_response_reader_rejects_disconnect_mid_frame(monkeypatch): + frame = encode_frame(make_response("request-1", ok=True)) + _response_reader(monkeypatch, [frame[:4], frame[4:8], b""]) + with pytest.raises(ProtocolError, match="disconnected"): + BrokerClient._read_response(object(), "request-1") diff --git a/agent-windows/tests/test_broker_service.py b/agent-windows/tests/test_broker_service.py index 30bef43..b9b475a 100644 --- a/agent-windows/tests/test_broker_service.py +++ b/agent-windows/tests/test_broker_service.py @@ -83,14 +83,26 @@ def test_parent_pause_reaches_capture_and_blocks_submit(tmp_path) -> None: assert config["paused"] is True assert config["capture_allowed"] is False assert config["policy_generation"] == "generation-2" + assert handler.applied_policy_generation() == "generation-2" assert "device_token" not in config assert handler.handle_message(make_request("status"))["payload"]["paused"] is True restarted = _handler(tmp_path) assert restarted.handle_message(make_request("status"))["payload"]["paused"] is True + assert restarted.applied_policy_generation() == "generation-2" restarted.set_capture_config({"paused": False, "capture_allowed": True}) assert restarted.handle_message(make_request("status"))["payload"]["paused"] is False +def test_failed_capture_policy_persistence_cannot_ack_generation(tmp_path, monkeypatch) -> None: + from src import broker_service + + handler = _handler(tmp_path) + monkeypatch.setattr(broker_service.os, "replace", lambda *_args: (_ for _ in ()).throw(OSError("locked"))) + handler.set_capture_config({"capture_allowed": False, "policy_generation": "generation-3"}) + assert handler.applied_policy_generation() is None + assert handler.handle_message(make_request("capture_config"))["payload"]["capture_allowed"] is False + + def test_broker_migrates_legacy_credentials_to_secure_path(tmp_path) -> None: handler = _handler(tmp_path) save_credentials("dev-1", "secret-token", "http://127.0.0.1:8787", handler.legacy_credential_path) diff --git a/agent-windows/tests/test_sender_retry.py b/agent-windows/tests/test_sender_retry.py index 3e73607..c686e65 100644 --- a/agent-windows/tests/test_sender_retry.py +++ b/agent-windows/tests/test_sender_retry.py @@ -34,3 +34,17 @@ async def test_sender_discards_permanent_payload_errors(): assert client.calls == 1 assert q.empty() + + +@pytest.mark.asyncio +async def test_sender_returns_on_unauthorized_without_losing_queued_frame(): + q: asyncio.Queue = asyncio.Queue(maxsize=2) + await q.put({"image_bytes": b"synthetic-frame", "app_name": "ExampleApp"}) + client = RejectingClient(401) + + await asyncio.wait_for(screenshot_sender_loop(client, q), timeout=1) + + assert client.calls == 1 + assert q.qsize() == 1 + assert (await q.get())["image_bytes"] == b"synthetic-frame" + q.task_done() From b1083721d84b4eb22bb7bca85776c93df04d1207 Mon Sep 17 00:00:00 2001 From: Trilobyte Date: Sat, 31 Oct 2026 10:04:36 -0400 Subject: [PATCH 13/29] fix: gate unqualified Windows enforcement by default --- backend/app/api/alerts.py | 3 +++ backend/app/settings.py | 3 +++ backend/tests/test_readiness_apis.py | 9 +++++++++ dashboard/src/pages/AlertDetail.tsx | 4 ++-- 4 files changed, 17 insertions(+), 2 deletions(-) diff --git a/backend/app/api/alerts.py b/backend/app/api/alerts.py index 4a38aed..c8a5bc4 100644 --- a/backend/app/api/alerts.py +++ b/backend/app/api/alerts.py @@ -11,6 +11,7 @@ from sqlalchemy.orm import Session from ulid import ULID +from app import settings as settings_mod from app.api.deps import current_user, get_db_dep, require_recent_auth from app.db.models import Alert, DeviceCommand, Event, RiskResult, User from app.services import encryption, purge @@ -403,6 +404,8 @@ def create_confirmed_action( raise HTTPException(404, "Alert not found") if not alert.device_id: raise HTTPException(409, "This alert is not linked to a child device") + if req.action in {"pause_app", "block_domain"} and not settings_mod.settings.device_enforcement_enabled: + raise HTTPException(409, "Device enforcement is not enabled on this installation") payload, preview = _action_payload(req) if req.confirmed_preview != preview: raise HTTPException(409, {"code": "confirmation_required", "preview": preview}) diff --git a/backend/app/settings.py b/backend/app/settings.py index 11c0e11..650d995 100644 --- a/backend/app/settings.py +++ b/backend/app/settings.py @@ -117,6 +117,9 @@ class Settings(BaseSettings): pending_frame_max_age_seconds: int = 600 pending_replay_max_frames: int = 50 guardian_review_enabled: bool = False + # App suspension and hosts-file edits need Windows crash/reboot/undo + # qualification before they are offered to ordinary families. + device_enforcement_enabled: bool = False guardian_review_provider: str = "mock" guardian_review_model: str = "gpt-5.6" guardian_review_codex_model: str = "gpt-5.6-sol" diff --git a/backend/tests/test_readiness_apis.py b/backend/tests/test_readiness_apis.py index a5409dc..fd33595 100644 --- a/backend/tests/test_readiness_apis.py +++ b/backend/tests/test_readiness_apis.py @@ -247,6 +247,7 @@ def test_alert_detail_includes_feed_context(monkeypatch, tmp_path): def test_alert_actions_are_effective_or_explicitly_unsupported(monkeypatch, tmp_path): client = _client(monkeypatch, tmp_path) + from app import settings as settings_mod from app.db.session import get_sessionmaker notice_version = client.get("/api/consent").json()["notice_version"] @@ -286,6 +287,14 @@ def test_alert_actions_are_effective_or_explicitly_unsupported(monkeypatch, tmp_ ) assert unsupported.status_code == 409 + gated = client.post( + "/api/alerts/action-alert/actions", + json={"action": "pause_app", "target": r"C:\Games\Chat.exe", "duration_seconds": 900}, + ) + assert gated.status_code == 409 + assert "not enabled" in gated.json()["detail"] + settings_mod.settings.device_enforcement_enabled = True + preview = client.post( "/api/alerts/action-alert/actions", json={ diff --git a/dashboard/src/pages/AlertDetail.tsx b/dashboard/src/pages/AlertDetail.tsx index 56db43d..26518f9 100644 --- a/dashboard/src/pages/AlertDetail.tsx +++ b/dashboard/src/pages/AlertDetail.tsx @@ -181,9 +181,9 @@ export default function AlertDetail() {

Parent-confirmed device action

-

GuardianNode never takes these actions from a classifier result. You will see and confirm the exact command first.

+

GuardianNode never takes these actions from a classifier result. You will see and confirm the exact command first. App pausing and domain blocking are unavailable while their Windows recovery behavior is being qualified.

- + {enforcementAction === "pause_app" && } {enforcementAction === "show_child_prompt" &&