diff --git a/.gitleaks.toml b/.gitleaks.toml index 3d6d031..b708a63 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -39,3 +39,18 @@ ever existed. Matched by `generic-api-key` because the fixtures are named `secre Keyed on the literal and nothing else — the file is still scanned for everything else. """ regexes = ['''AKIAABCDEFGHIJKLMNOP'''] + +[[allowlists]] +description = """ +A high-entropy nonsense string in the keyword-proximity tests +(packages/detect/tests/test_signals.py, introduced in e6b1317). It is the *subject* +of those tests rather than a credential: they assert that `oauth` clipped by a window +edge does not count as the keyword `auth`, which needs a token entropic enough for +`generic-api-key` to match — so the fixture and the finding have the same cause. + +Allowlisted here rather than with an inline `# gitleaks:allow` because e6b1317 is +already in history and the CI job scans all of it; editing the file today would not +change what that commit contains. Keyed on the literal, so the file is still scanned +for everything else. +""" +regexes = ['''8f3Kd0aQ2mVx7Lp9Zr4Ts6'''] diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 6022e63..040bc7f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,3 +1,11 @@ +# Vendored frontend assets and the brand marks are byte-pinned: static/assets.lock.json +# records a sha384 of each, base.html emits it as an `integrity=` attribute, and +# apps/api/tests/test_web_invariants.py checks the two agree. A whitespace hook that +# "fixes" a minified bundle changes those bytes, so the browser blocks the script and +# the page silently loses its behaviour — with a diff that looks like nothing happened. +# Leave them exactly as web/vendor_assets.py wrote them. +exclude: ^apps/api/src/iceberg_api/web/static/(js/vendor|css/vendor|fonts|img)/ + repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v6.0.0 diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2bec33f..ab867b6 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -160,8 +160,9 @@ connector credentials, full audit trail on finding state changes, and RBAC on ev - **M1 — Control plane MVP:** auth, sources/schedules, scan orchestration, findings/triage. - **M2 — Detection + Confluence:** detection engine, connector framework, Confluence connector, engine worker. -- **M3 — Web UI:** HTMX/Alpine screens. -- **M4 — Notifications & prod deploy:** notifications, Helm, hardening. +- **M3 — Web UI:** HTMX/Alpine screens, served by the API at the application root and driving the + same route handlers the JSON API exposes ([`docs/web.md`](./docs/web.md)). +- **M4 — Notifications & prod deploy:** notification dispatch, Helm, hardening. Non-goals for MVP: Jira/SMB connectors, incremental/delta scanning, image OCR, external ticket creation, multi-tenancy. diff --git a/CLAUDE.md b/CLAUDE.md index 5e0228a..0e753a8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,12 +8,13 @@ IcebergSST — a secret-scanning platform for non-git enterprise sources (Conflu shares). See [`ARCHITECTURE.md`](./ARCHITECTURE.md) for the design spec and [`docs/adr/`](./docs/adr/) for decision rationale. -**Where the code is:** M0–M2 are built. `packages/core` (config, DB session, secret store, +**Where the code is:** M0–M3 are built. `packages/core` (config, DB session, secret store, redaction, fingerprinting, the SQLModel schema), `packages/detect` (rule packs, the detection engine), `packages/connectors` (Confluence, extraction, the sandbox), `apps/api` (auth, sources, -scans, findings, engine-facing routes, scheduler/maintenance), `apps/engine` (the Dramatiq -worker), Alembic, and the docker-compose stack all exist and are tested. The HTMX/Alpine web UI -(M3) and notifications/Helm (M4) are still to come — `docs/backlog.md` tracks those. +scans, findings, notification channels, engine-facing routes, scheduler/maintenance, and the +server-rendered web console under `apps/api/src/iceberg_api/web/`), `apps/engine` (the Dramatiq +worker), Alembic, and the docker-compose stack all exist and are tested. Notification *dispatch* +and Helm (M4) are still to come — `docs/backlog.md` tracks those. ## Non-negotiable invariants @@ -28,8 +29,13 @@ change the corresponding ADR: 3. **Detection logic in code, tuning in data.** Rules live in versioned rule packs (`packages/detect`). Suppressions/allowlists live in the DB and are analyst-editable. 4. **API-first.** The FastAPI OpenAPI schema is the contract; the HTMX/Alpine UI is just another - client of the same routes. + client of the same routes. Concretely: a web route resolves the RBAC dependency its API + counterpart declares and then calls that handler as a function. Nothing under + `iceberg_api.web` queries the database (see `docs/web.md`). 5. **Single-org.** No `tenant_id`, no row-level tenancy. +6. **No inline script or style in a template.** The console runs under + `script-src 'self'` with no `'unsafe-inline'`/`'unsafe-eval'`; Alpine components go in + `static/js/tags.js` and server data travels through JSON islands. ## Stack @@ -53,6 +59,11 @@ Prefer extending these over adding prose: configuration, no master key, and stay scalable; `.env.example` documents every variable compose interpolates. - `apps/api/tests/test_migrations.py` — migrations apply, match the metadata, and reverse. +- `apps/api/tests/test_web_invariants.py` — no `iceberg_api.web` module imports the ORM or builds a + query, every mutating browser route is CSRF-protected, no HTML endpoint appears in the OpenAPI + document, and every vendored asset matches its `assets.lock.json` integrity. +- `apps/api/tests/test_web_shell.py` — no template carries an inline ` +
+ +
+ + +
+ + + + + + + + + + + +
+ +
+
+ + + +
+ {% if channels %} +
+ + + + + + {% for channel in channels %} + + + + + + + + + + {% endfor %} + +
NameTypeDestinationFilterSecretStateActions
{{ channel.name }}{{ channel.type.value }} + {% if channel.type.value == 'webhook' %} + {{ channel.config.get('url', '—') }} + {% else %} + {{ channel.config.get('recipients', []) | join(', ') or '—' }} + {% endif %} + + {{ (channel.event_filter.min_severity.value ~ '+') if channel.event_filter.min_severity else 'all' }} + + {% if channel.has_secret %} + sealed + {% else %} + none + {% endif %} + + {{ 'enabled' if channel.enabled else 'disabled' }} + + + + + + +
+
+ {% else %} +
+
No channels
+

Nothing leaves this deployment. Findings are visible in the console and through the API only.

+
+ {% endif %} +
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/admin/engines.html b/apps/api/src/iceberg_api/web/templates/admin/engines.html new file mode 100644 index 0000000..568870f --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/admin/engines.html @@ -0,0 +1,133 @@ +{% extends "base.html" %} +{% block title %}Engines · {{ app_name }}{% endblock %} +{% block crumb %}Engines{% endblock %} + +{% block content %} +
+
+
Administration
+

Engine fleet

+
+
+
+ +

+ Engines hold no database credentials. They lease tasks over the API, receive the source credential + and the fingerprint pepper for the duration of one task, and post redacted results back. An engine + that stops heartbeating loses its leases, and the API redispatches its work. +

+ +{% if error %} + +{% endif %} + +{% if credential %} +{#- The one moment this value exists in a response. The API stores only a + SHA-256 of it and cannot show it again, which is why this is rendered + in place rather than redirected to. -#} +
+
Copy this now — it is not shown again
+

Token for {{ credential.name }}

+ 
{{ credential.token }}
+
+ + ICEBERG_ENGINE_ID={{ credential.engine_id }} +
+

+ Set ICEBERG_ENGINE_TOKEN and ICEBERG_ENGINE_ID on the engine. Only its + SHA-256 hash is stored here; re-enrolling the same name rotates the token and the old one stops + working immediately. +

+
+{% endif %} + +
+

Fleet

+ {% if engines %} +
+ + + + {% for engine in engines %} + + + + + + + + {% endfor %} + +
NameStatusVersionLast heartbeatEnrolled
{{ engine.name }} + {{ engine.status.value }} + {{ engine.version or '—' }} + {{ engine.last_heartbeat_at | ago }} + · {{ engine.last_heartbeat_at | dt }} + {{ engine.created_at | dt }}
+
+ {% else %} +
+
No engines enrolled
+

Nothing will scan until an engine has a token. Enrol one below, or mint the first from the + CLI with python -m iceberg_api mint-engine-token --name engine-1.

+
+ {% endif %} +
+ +
+
+

Enrol an engine

+

+ An engine cannot enrol itself — otherwise anyone who could reach the API could join the fleet + and start leasing tasks, credentials included. Re-using an existing name rotates that engine's + token, which is how you replace one you think has leaked. +

+
+ + +
+ +
+
+
+ +
+
Detection surface
+ {% if rules.fleet_consistent %} +
+
Consistent. Every live engine reports the same rule pack.
+
+ {% else %} +
+
+ Two rule packs are in force. Normal during a rolling deploy — each finding + records the version that produced it. +
+
+ {% endif %} +
+ {% for version in rules.rulepack_versions %} +
{{ version.version }}
+
{{ version.engine_count }} engine(s) · {{ version.rule_count }} rules
+ {% endfor %} +
Threshold
{{ rules.confidence_threshold }}
+
+ {% if rules.engines_without_a_rulepack %} +

+ Not reporting a pack: {{ rules.engines_without_a_rulepack | join(', ') }} +

+ {% endif %} + + All rules + + +
+
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/admin/rules.html b/apps/api/src/iceberg_api/web/templates/admin/rules.html new file mode 100644 index 0000000..870d764 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/admin/rules.html @@ -0,0 +1,101 @@ +{% extends "base.html" %} +{% block title %}Rules · {{ app_name }}{% endblock %} +{% block crumb %}Rules{% endblock %} + +{% block content %} +
+
+
Configuration
+

Detection surface

+
+
+
+ +

+ What the live fleet is actually detecting, reported by the engines themselves rather than declared + by the API. Rule packs ship inside engine images, so a list published here would describe what the + fleet is supposed to be running — which is the thing this page exists to check. + Rules are metadata only: never a pattern. +

+ +{% if not rules.fleet_consistent %} +
+
+ The fleet disagrees. More than one rule-pack version is in force — normal + mid-deploy. Every finding records the version that produced it, so results stay reproducible + either way. +
+
+{% endif %} + +
+
+
Rules in force
+
{{ rules.rules | length }}
+
+
+
Pack versions
+
{{ rules.rulepack_versions | length }}
+
+
+
Confidence threshold
+
{{ rules.confidence_threshold }}
+
matches below this are dropped
+
+
+
Silent engines
+
{{ rules.engines_without_a_rulepack | length }}
+
registered, no pack reported
+
+
+ +{% if rules.rulepack_versions %} +
+

Packs

+
+ + + + {% for version in rules.rulepack_versions %} + + + + + + + {% endfor %} + +
VersionEnginesRulesRunning it
{{ version.version }}{{ version.engine_count }}{{ version.rule_count }}{{ version.engines | join(', ') }}
+
+
+{% endif %} + +
+

Rules

+ {% if rules.rules %} +
+ + + + {% for rule in rules.rules %} + + + + + + + + {% endfor %} + +
RuleSeverityDescriptionDefined inFindings
{{ rule.id }}{{ rule.severity.value }}{{ rule.description }}{{ rule.rulepack_versions | join(', ') }} + Open +
+
+ {% else %} +
+
No engine has reported a rule pack
+

Enrol an engine and let it heartbeat once; the pack it loaded shows up here.

+
+ {% endif %} +
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/admin/users.html b/apps/api/src/iceberg_api/web/templates/admin/users.html new file mode 100644 index 0000000..e925713 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/admin/users.html @@ -0,0 +1,100 @@ +{% extends "base.html" %} +{% block title %}Users · {{ app_name }}{% endblock %} +{% block crumb %}Users{% endblock %} + +{% block content %} +
+
+
Administration
+

Users & roles

+
+
+
+ +

+ Accounts are created by the identity provider on first sign-in and land as viewer + — somebody has an account because the IdP knows them, not because anyone decided what they may do. + Roles are assigned here and never taken from a token claim, so no IdP change can promote anyone. + Every change below writes an audit event. +

+ +{% if error %} + +{% endif %} + +
+
+ You cannot change your own account. It closes the obvious escalation and removes + every path to locking the last administrator out. A legitimate change to your own role is one + another admin makes for you. +
+
+ +
+ {% if users %} +
+ + + + + + {% for account in users %} + + + + + + + + + {% endfor %} + +
PersonSubjectRoleStateLast sign-inChange
+ {{ account.display_name }} +
{{ account.email }}
+ 		{{ account.oidc_subject }}{{ account.role.value }} + {% if account.disabled %} + disabled + {% else %} + active + {% endif %} + {{ account.last_login_at | dt }} + {% if account.id == user.id %} + that is you + {% else %} +
+ + + + +
+ {% endif %} +
+
+ {% else %} +
+
No accounts yet
+

The first person to sign in through the identity provider gets one.

+
+ {% endif %} +
+ +{% if next_cursor %} +
+ Showing one page + + Next page + + +
+{% endif %} +{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/base.html b/apps/api/src/iceberg_api/web/templates/base.html new file mode 100644 index 0000000..543dc1f --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/base.html @@ -0,0 +1,117 @@ +{#- The authenticated shell: dark rail | (breadcrumb topbar / light canvas). + + Every asset below is first-party and version-pinned, with an `integrity` + hash from static/assets.lock.json. There is no inline + + + + +
+ {% include "_icons.html" %} + + {#- Every hx- request carries the session's CSRF token. Set once here rather + than on each element: the API rejects a mutation without it (by design), + and a token added per-form is one somebody eventually forgets. -#} +
+ {% if user %} + {%- macro nav_link(href, icon, label, active) -%} + + + {{ label }} + + {%- endmacro -%} + + + {% endif %} + +
+ {% if user %} +
+
+ {{ app_name }} + / + {% block crumb %}Overview{% endblock %} +
+
{% block topbar_tools %}{% endblock %}
+
+ {% endif %} +
+
+ {% block content %}{% endblock %} +
+
+
+
+ + diff --git a/apps/api/src/iceberg_api/web/templates/error.html b/apps/api/src/iceberg_api/web/templates/error.html new file mode 100644 index 0000000..b2be96a --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/error.html @@ -0,0 +1,35 @@ +{#- A failure a person can act on. Rendered without the rail: the two statuses + that reach it most often are 403 (this role may not be here) and 404, and + both can happen before there is a session to build a rail from. -#} + + + + + + {{ status_code }} · {{ app_name }} + + + + + +
+
+
+
+ + IcebergSST +
+
Error {{ status_code }}
+

{{ title }}

+
+

{{ explanation }}

+ {% if detail %} +

{{ detail }}

+ {% endif %} +
+ Back to the overview +
+
+
+ + diff --git a/apps/api/src/iceberg_api/web/templates/findings/detail.html b/apps/api/src/iceberg_api/web/templates/findings/detail.html new file mode 100644 index 0000000..4f51353 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/findings/detail.html @@ -0,0 +1,112 @@ +{% extends "base.html" %} +{% block title %}{{ finding.rule_id }} · Findings · {{ app_name }}{% endblock %} +{% block crumb %}{{ finding.rule_id }}{% endblock %} + +{% block content %} +← Findings + +
+
+
Finding
+

{{ finding.rule_id }}

+
+ {{ finding.severity.value }} + {{ finding.state.value | humanize }} + {% if finding.resolution %}{{ finding.resolution.value }} resolution{% endif %} + {% if finding.suppressed_at %}suppressed{% endif %} +
+
+
+ +{% if finding.suppressed_at %} +
+
+ Suppressed {{ finding.suppressed_at | ago }}. + It is recorded, not discarded — out of the active view until the suppression lapses or is + deleted, at which point it comes back with an event saying why. +
+
+{% endif %} + +
+
+
+

Evidence

+
+
Redacted snippet
+ {#- Masked inside the engine before it left the worker (ADR 0004). The + database has never held the plaintext, so there is nothing here to + reveal — and the string is autoescaped because it came out of + somebody else's Confluence. -#} + 
{{ finding.redacted_snippet }}
+

+ The secret itself was never transmitted or stored — only this mask and a peppered hash, + which is what makes the finding identifiable across re-scans without being reversible. +

+
+
+ +
+

Where it is

+
+
+ {% for key, value in finding.resource_locator.items() %} +
{{ key | humanize }}
+
+ {% if key in ('url', 'link') and value is string and (value.startswith('https://') or value.startswith('http://')) %} + {#- rel=noopener: the target is an operator-supplied instance, and a + new tab must not get a handle on this one. -#} + {{ value }} + + + {% else %} + {{ value }} + {% endif %} +
+ {% endfor %} +
+
+
+ + {% include "partials/triage.html" %} +
+ +
+
+
Record
+
+
Id
{{ finding.id }}
+
Fingerprint
{{ finding.fingerprint }}
+
Rule pack
{{ finding.rulepack_version }}
+
Confidence
{{ '%.2f' % finding.confidence if finding.confidence is not none else 'not scored' }}
+
Entropy
{{ '%.2f' % finding.entropy if finding.entropy is not none else '—' }}
+
Source
{{ finding.source_id | short }}
+
First seen
{{ finding.created_at | dt }}
+
Last seen
{{ finding.updated_at | dt }}
+
Assignee
+
{{ user_names.get(finding.assignee_id, finding.assignee_id | short) if finding.assignee_id else 'unassigned' }}
+
+ {% if finding.notes %} +
+
Notes
+

{{ finding.notes }}

+ {% endif %} +
+ + {% if role in ('analyst', 'admin') %} +
+
Suppress
+

+ If this exact match is known-benign, a fingerprint suppression hides it here and at every + future ingest — and records why. +

+ + Create a suppression + + +
+ {% endif %} +
+
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/findings/list.html b/apps/api/src/iceberg_api/web/templates/findings/list.html new file mode 100644 index 0000000..1e08514 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/findings/list.html @@ -0,0 +1,128 @@ +{% extends "base.html" %} +{% block title %}Findings · {{ app_name }}{% endblock %} +{% block crumb %}Findings{% endblock %} +{% block page_class %}page-wide{% endblock %} + +{% block content %} +
+
+
Workspace
+

Findings

+
+
+
+ Open queue + Suppressed +
+
+ +

+ Every filter below is part of the URL and is sent to the API as written — nothing is hidden + behind your back, so the number of rows here always reconciles with the query above them. +

+ +
+ + + + + + + + Clear +
+ +
+ {% if findings %} +
+ + + + + + + + + {% for finding in findings %} + + + + + + + + + + {% endfor %} + +
SeverityStateRuleSourceWhereAssigneeFirst seen
{{ finding.severity.value }} + {{ finding.state.value | humanize }} + {% if finding.suppressed_at %}suppressed{% endif %} + {{ finding.rule_id }}{{ source_names.get(finding.source_id, finding.source_id | short) }} + {{ finding.resource_locator.get('path') or finding.resource_locator.get('url') or finding.resource_locator.get('title') or '—' }} + {{ finding.assignee_id | short if finding.assignee_id else '—' }}{{ finding.created_at | dt }}
+
+ {% else %} +
+
Nothing matches this query
+

Widen the filters, or clear them to see everything recorded.

+
+ {% endif %} +
+ +{% if next_cursor %} +
+ Keyset paging — no skipped or repeated rows + + Next page + + +
+{% endif %} +{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/login.html b/apps/api/src/iceberg_api/web/templates/login.html new file mode 100644 index 0000000..4a7bfcc --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/login.html @@ -0,0 +1,46 @@ +{#- The one page an anonymous visitor may see. + + No password field: authentication is OIDC only (ADR 0005), so the single + control here starts the authorization-code flow. `next` is carried through to + /auth/login, which reduces it to a local path before it is ever used as a + redirect target. +-#} + + + + + + Sign in · {{ app_name }} + + + + + +
+
+
+
+ + IcebergSST +
+
Secret scanning
+

Sign in

+
+

+ This console is authenticated by your organisation's identity provider. + A first-time sign-in lands as a viewer; an administrator grants anything more. +

+ {% if error %} + + {% endif %} + + Continue with single sign-on + + + {% include "_icons.html" %} +
+
+ + diff --git a/apps/api/src/iceberg_api/web/templates/overview.html b/apps/api/src/iceberg_api/web/templates/overview.html new file mode 100644 index 0000000..2489074 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/overview.html @@ -0,0 +1,137 @@ +{% extends "base.html" %} +{% block title %}Overview · {{ app_name }}{% endblock %} +{% block crumb %}Overview{% endblock %} + +{% block content %} +
+
+
Secret scanning
+

Overview

+
+
+ {% if role in ('analyst', 'admin') %} +
+ Run a scan +
+ {% endif %} +
+ +
+ +
Critical, open
+
{{ critical_count }}{{ '+' if critical_capped }}
+
Highest severity, not suppressed
+ + +
Open findings
+
{{ open_count }}{{ '+' if open_capped }}
+
{{ unassigned_count }} unassigned{{ ' (of the first page)' if open_capped }}
+ + +
Assigned to you
+
{{ mine_count }}
+
Waiting on your decision
+ + +
Sources
+
{{ source_count }}{{ '+' if source_capped }}
+
Configured for scanning
+ +
+ +{% if open_capped %} +
+
+ Counts are capped. The API returns at most {{ open_count }} rows per page, so + these tiles say “{{ open_count }}+” rather than guess at the total. The tables behind them page + through everything. +
+
+{% endif %} + +{% if stale_engines %} +
+
+ {{ stale_engines | length }} engine(s) are not active. + Queued scan tasks will wait until one comes back — check the fleet. +
+
+{% endif %} + +
+
+
+

Newest open findings

+ All findings +
+ {% if recent_findings %} +
+ + + + + + {% for finding in recent_findings %} + + + + + + + + {% endfor %} + +
SeverityRuleSourceWhereSeen
{{ finding.severity.value }}{{ finding.rule_id }}{{ sources[finding.source_id].name if finding.source_id in sources else finding.source_id | short }}{{ finding.resource_locator.get('path') or finding.resource_locator.get('url') or '—' }}{{ finding.created_at | ago }}
+
+ {% else %} +
+
Nothing open
+

Either no scan has run yet, or every finding has been triaged. Both are good places to be.

+
+ {% endif %} +
+ +
+
+
+

Running now

+ All scans +
+ {% if active_scans %} +
+ {% for scan in active_scans %} + + + + {{ sources[scan.source_id].name if scan.source_id in sources else scan.source_id | short }} + {{ scan.status.value }} · started {{ scan.started_at | ago }} + + + {% endfor %} +
+ {% else %} +
No scan is running.
+ {% endif %} +
+ +
+

Latest scans

+ {% if recent_scans %} +
+ {% for scan in recent_scans %} + + {{ scan.status.value }} + + {{ sources[scan.source_id].name if scan.source_id in sources else scan.source_id | short }} + {{ scan.trigger.value }} · {{ scan.created_at | ago }} + + + {% endfor %} +
+ {% else %} +
No scans yet.
+ {% endif %} +
+
+
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/partials/connectivity.html b/apps/api/src/iceberg_api/web/templates/partials/connectivity.html new file mode 100644 index 0000000..7f3475d --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/partials/connectivity.html @@ -0,0 +1,28 @@ +{#- The connectivity-check result, swapped in beside the button that asked. + + Nothing from the source's response body appears here. The API reduces the + probe to a reachable flag, a status code, and a bounded explanation — the URL + is operator-supplied and the response is untrusted (docs/security.md + § Outbound requests). +-#} +
+ {% if error %} +
+
Could not test. {{ error }}
+
+ {% elif result.reachable %} +
+
+ Reachable. {{ result.detail }} + {% if result.status_code %} · HTTP {{ result.status_code }}{% endif %} +
+
+ {% else %} +
+
+ Not reachable. {{ result.detail }} + {% if result.status_code %} · HTTP {{ result.status_code }}{% endif %} +
+
+ {% endif %} +
diff --git a/apps/api/src/iceberg_api/web/templates/partials/scan_status.html b/apps/api/src/iceberg_api/web/templates/partials/scan_status.html new file mode 100644 index 0000000..012ed5e --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/partials/scan_status.html @@ -0,0 +1,128 @@ +{#- The live status region (#56). + + It polls *itself*: the `hx-trigger="every Ns"` lives on this fragment, and the + terminal version of the fragment omits it. So a scan that finishes stops the + polling by the act of rendering — the browser never needs to know which + statuses are terminal, and there is no timer to leak. +-#} +
+ + {% if message %} +
+
Nothing to cancel. {{ message }}
+
+ {% endif %} + +
+
+
+ {% if active %}{% endif %} + {{ scan.status.value }} + + {% if active %} + {{ task_finished }} of {{ task_total }} task(s) finished + {% elif scan.finished_at %} + finished {{ scan.finished_at | ago }} + {% else %} + not started + {% endif %} + +
+ {% if active and role in ('analyst', 'admin') %} +
+ + + + + +
+ {% endif %} +
+ +
+ +
+ + {% if scan.error %} + + {% endif %} + + {% if scan.status.value == 'partial' %} +
+
+ Partial. Some tasks failed, so this scan did not see the whole source — + findings it did not observe were deliberately not auto-resolved. +
+
+ {% elif scan.status.value == 'cancelled' %} +
+
+ Cancelled. A cancelled scan never reconciles: it saw part of the source, + and resolving findings on that basis would be wrong. +
+
+ {% endif %} + +
+
+
Units scanned
+
{{ scan.counts.get('units', 0) }}
+
{{ scan.counts.get('units_skipped', 0) }} skipped · {{ scan.counts.get('units_failed', 0) }} failed
+
+
+
Findings
+
{{ scan.counts.get('findings', 0) }}
+
{{ scan.counts.get('suppressed', 0) }} suppressed at ingest
+
+
+
New / reopened
+
{{ scan.counts.get('new', 0) }}
+
{{ scan.counts.get('reopened', 0) }} reopened
+
+
+
Auto-resolved
+
{{ scan.counts.get('resolved', 0) }}
+
no longer present at the source
+
+
+
+ +
+

Tasks

+ {% if tasks %} +
+ + + + {% for task in tasks %} + + + + + + + + + + {% endfor %} + +
KindStatusEngineAttemptsLeaseFinishedError
{{ task.kind.value }}{{ task.status.value }}{{ task.engine_id | short if task.engine_id else '—' }}{{ task.attempts }}{{ task.lease_expires_at | dt }}{{ task.finished_at | dt }}{{ task.error or '' }}
+
+ {% else %} +
No tasks yet — the discovery task has not been created.
+ {% endif %} +
+
diff --git a/apps/api/src/iceberg_api/web/templates/partials/source_form.html b/apps/api/src/iceberg_api/web/templates/partials/source_form.html new file mode 100644 index 0000000..1bb470f --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/partials/source_form.html @@ -0,0 +1,170 @@ +{#- The reference partial (#54). Everything the conventions describe is here: + + * one root element with a stable id, so the mutation can swap it whole; + * a `x-data` naming a component registered in tags.js — no behaviour inline; + * server state handed to Alpine through a JSON island, not an attribute; + * `hx-post` to a route that answers with *this same fragment* on failure, so + a rejected save leaves the analyst's typing on screen; + * a `hx-indicator` so a slow save looks like a slow save. + + The credential field is never populated. The API cannot return one and this + form must not imply that the value it shows is the one in force. +-#} +
+ + + {% if error %} + + {% endif %} + +
+ + +
+ + {% if not source %} + + {% endif %} +
+ + + + + + + {#- Bound to the computed value so Server/DC posts an empty email, which the + API reads as "omitted" — the flag that selects bearer auth. -#} + + + + +
+ Spaces +
+ + +
+
+ + Every space the credential can read. +
+
+ + + + + + {% if source and source.has_credential %} +
+ Credential +
+ stored + +
+
+ + + Leave this blank to keep the existing credential. There is no way to remove one: a source + that cannot authenticate cannot be scanned, so delete the source instead. + + +
+
+ {% else %} + + {% endif %} + + + +
+ + {% if source %} + Cancel + {% else %} + Cancel + {% endif %} +
+
+
diff --git a/apps/api/src/iceberg_api/web/templates/partials/triage.html b/apps/api/src/iceberg_api/web/templates/partials/triage.html new file mode 100644 index 0000000..7b6e8a3 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/partials/triage.html @@ -0,0 +1,125 @@ +{#- The triage panel and the finding's history, swapped together. + + They are one fragment because they are one answer: a state change writes a + FindingEvent, and showing the new state without the entry that explains it + would make the trail look like it lagged. `PATCH /findings/{id}` returns the + detail shape *with* its events for the same reason. +-#} +
+ + + {% if error %} + + {% endif %} + + {% if role in ('analyst', 'admin') %} +
+
Triage
+
+ + + + +
+
+
Reopening clears the resolution, so this finding will not keep an automatic one.
+
+
+ + + + + +
+ + Change the state or add a comment. +
+
+
+ {% else %} +
+
Triage
+

+ Triage is an analyst decision — it has to have a name attached to it. Your role can read + findings but not change them. +

+
+ {% endif %} + +
+

History

+
+ {% if finding.events %} +
    + {% for event in finding.events %} +
  1. +
    + {{ event.kind.value | humanize }} + + {#- A null actor is the system: reconciliation auto-resolving, or + ingest applying a suppression. Naming it is the point. -#} + {{ user_names.get(event.actor_id, 'system') if event.actor_id else 'system' }} + + {{ event.created_at | dt }} +
    + {% if event.from_value or event.to_value %} +
    + {{ event.from_value or '—' }} → {{ event.to_value or '—' }} +
    + {% endif %} + {% if event.comment %} +
    {{ event.comment }}
    + {% endif %} +
    2. + {% endfor %} +
+ {% else %} +

Nothing has happened to this finding since it was ingested.

+ {% endif %} +
+
+
diff --git a/apps/api/src/iceberg_api/web/templates/scans/detail.html b/apps/api/src/iceberg_api/web/templates/scans/detail.html new file mode 100644 index 0000000..a030b3c --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/scans/detail.html @@ -0,0 +1,38 @@ +{% extends "base.html" %} +{% block title %}Scan {{ scan.id | short }} · {{ app_name }}{% endblock %} +{% block crumb %}Scan {{ scan.id | short }}{% endblock %} + +{% block content %} +← All scans + +
+
+
{{ scan.trigger.value }} scan of
+

{{ source.name }}

+
+
+
+ + Findings from this source + + +
+
+ +
+ {#- The status region owns its own polling; see partials/scan_status.html. -#} + {% include "partials/scan_status.html" %} +
+ +
+
Scan record
+
+
Id
{{ scan.id }}
+
Trigger
{{ scan.trigger.value }}
+
Rule pack
{{ scan.rulepack_version or 'not reported yet' }}
+
Created
{{ scan.created_at | dt }}
+
Started
{{ scan.started_at | dt }}
+
Finished
{{ scan.finished_at | dt }}
+
+
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/scans/list.html b/apps/api/src/iceberg_api/web/templates/scans/list.html new file mode 100644 index 0000000..44e107a --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/scans/list.html @@ -0,0 +1,86 @@ +{% extends "base.html" %} +{% block title %}Scans · {{ app_name }}{% endblock %} +{% block crumb %}Scans{% endblock %} + +{% block content %} +
+
+
Workspace
+

Scans

+
+
+
+ +{#- A plain GET form. The filters are in the URL, so a filtered view can be + bookmarked, linked in a ticket, and reloaded — none of which a client-side + filter gives you. -#} +
+ + + + Clear +
+ +
+ {% if scans %} +
+ + + + + + {% for scan in scans %} + + + + + + + + + + {% endfor %} + +
StatusSourceTriggerFindingsRule packStartedFinished
+ + {{ scan.status.value }} + + + {{ source_names.get(scan.source_id, scan.source_id | short) }} + {{ scan.trigger.value }}{{ scan.counts.get('findings', 0) }} / {{ scan.counts.get('units', 0) }} units{{ scan.rulepack_version or '—' }}{{ scan.started_at | dt }}{{ scan.finished_at | dt }}
+
+ {% else %} +
+
No scans match
+

Launch one from a source, or clear the filters above.

+
+ {% endif %} +
+ +{% if next_cursor %} +
+ Showing one page + + Next page + + +
+{% endif %} +{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/schedules/list.html b/apps/api/src/iceberg_api/web/templates/schedules/list.html new file mode 100644 index 0000000..a5e5a87 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/schedules/list.html @@ -0,0 +1,144 @@ +{% extends "base.html" %} +{% block title %}Schedules · {{ app_name }}{% endblock %} +{% block crumb %}Schedules{% endblock %} + +{% block content %} +
+
+
Configuration
+

Schedules

+
+
+
+ +

+ A cron cadence attached to a source. Every API replica runs the scheduler; a Postgres advisory + lock decides which one actually fires, so a schedule never double-launches. +

+ +{% if error %} + +{% endif %} + +{% if role == 'admin' %} +
+ + +
+ +
+ + + +
+ {% for expression, label in presets %} + + {% endfor %} +
+ +
+ +
+
+
+
+{% endif %} + +
+ + + Clear +
+ +
+ {% if schedules %} +
+ + + + {% if role == 'admin' %}{% endif %} + + + {% for schedule in schedules %} + + + + + + + {% if role == 'admin' %} + + {% endif %} + + {% endfor %} + +
SourceCronStateNext runLast runActions
{{ source_names.get(schedule.source_id, schedule.source_id | short) }} + {% if role == 'admin' %} +
+ + + + +
+ {% else %} + {{ schedule.cron }} + {% endif %} + 		+ {{ 'enabled' if schedule.enabled else 'disabled' }} + {{ schedule.next_run_at | dt }}{{ schedule.last_run_at | dt }} + + + + + +
+
+ {% else %} +
+
No schedules
+

Every source here is scanned on demand only.

+
+ {% endif %} +
+ +{% if next_cursor %} +
+ Showing one page + + Next page + + +
+{% endif %} +{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/sources/detail.html b/apps/api/src/iceberg_api/web/templates/sources/detail.html new file mode 100644 index 0000000..f31d85f --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/sources/detail.html @@ -0,0 +1,164 @@ +{% extends "base.html" %} +{% block title %}{{ source.name }} · Sources · {{ app_name }}{% endblock %} +{% block crumb %}{{ source.name }}{% endblock %} + +{% block content %} +← All sources + +
+
+
{{ source.type.value }} source
+

{{ source.name }}

+
+
+
+ {% if role in ('analyst', 'admin') %} + {#- Launching answers 202 and redirects to the new scan, where the live + status view takes over (#56). Disabled sources cannot be scanned, and the + API says so with a 409 — the button reflects that rather than earning one. -#} + + {% endif %} + {% if role == 'admin' %} + + {% endif %} +
+
+ +
+ +{% if not source.enabled %} +
+
This source is disabled. Scheduled and manual scans are both refused until it is enabled.
+
+{% endif %} +{% if not source.has_credential %} +
+
No credential stored. A scan will fail to authenticate. Add one below.
+
+{% endif %} + +
+
+
+

Recent scans

+ {% if scans %} +
+ + + + {% for scan in scans %} + + + + + + + + {% endfor %} + +
StatusTriggerFindingsStartedFinished
{{ scan.status.value }}{{ scan.trigger.value }}{{ scan.counts.get('findings', 0) }} found / {{ scan.counts.get('units', 0) }} scanned{{ scan.started_at | dt }}{{ scan.finished_at | dt }}
+
+ {% else %} +
This source has never been scanned.
+ {% endif %} +
+ + {% if role == 'admin' %} +
+

Configuration

+ {% with source = form.source, connection = form.connection, types = form.types, + island = form.island, error = none %} + {% include "partials/source_form.html" %} + {% endwith %} +
+ {% endif %} +
+ +
+
+
Details
+
+
Id
{{ source.id }}
+
Type
{{ source.type.value }}
+
Base URL
{{ source.connection.get('base_url', '—') }}
+
Auth
{{ 'Cloud (email + token)' if source.connection.get('email') else 'Server/DC (bearer token)' }}
+
Spaces
+
+ {% if source.connection.get('spaces') %} + {% for space in source.connection.spaces %}{{ space }} {% endfor %} + {% else %} + every readable space + {% endif %} +
+
Comments
{{ 'scanned' if source.connection.get('include_comments', True) else 'skipped' }}
+
Attachments
{{ 'scanned' if source.connection.get('include_attachments', True) else 'skipped' }}
+
Personal
{{ 'included' if source.connection.get('include_personal_spaces', False) else 'excluded' }}
+
Created
{{ source.created_at | dt }}
+
Updated
{{ source.updated_at | dt }}
+
+
+ +
+
+

Schedules

+ Manage +
+ {% if schedules %} +
+ {% for schedule in schedules %} +
+ {{ 'on' if schedule.enabled else 'off' }} + + {{ schedule.cron }} + next {{ schedule.next_run_at | dt }} + +
+ {% endfor %} +
+ {% else %} +
No schedule — this source is scanned on demand only.
+ {% endif %} +
+ +
+
Findings
+ + Open findings from this source + + +
+ + {% if role == 'admin' %} +
+
Danger zone
+

+ Deleting a source deletes its scans, findings, and schedules. Triage decisions go with them. +

+ +
+ + +
+
+ {% endif %} +
+
+{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/sources/list.html b/apps/api/src/iceberg_api/web/templates/sources/list.html new file mode 100644 index 0000000..88b76d8 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/sources/list.html @@ -0,0 +1,97 @@ +{% extends "base.html" %} +{% block title %}Sources · {{ app_name }}{% endblock %} +{% block crumb %}Sources{% endblock %} + +{% block content %} +
+
+
Configuration
+

Sources

+
+
+
+ +

+ Where IcebergSST looks. Each source carries its own credential, sealed through the secret store — + no response ever returns it, and rotating one is an audited action separate from editing the source. +

+ +{% if role == 'admin' %} +{#- Creating is demoted behind a disclosure so the list is the page, not the + form. `data-open` lets the server decide the initial state. -#} +
+ +
+ {#- The partial reads these names from the surrounding context; `with` is + what supplies them when a page embeds it rather than a route rendering + it on its own. -#} + {% with source = form.source, connection = form.connection, types = form.types, + island = form.island, error = none %} + {% include "partials/source_form.html" %} + {% endwith %} +
+
+{% endif %} + +
+ {% if sources %} +
+ + + + + + + + {% for source in sources %} + + + + + + + + + {% endfor %} + +
NameTypeWhereCredentialStateAdded
{{ source.name }}{{ source.type.value }}{{ source.connection.get('base_url', '—') }} + {% if source.has_credential %} + stored + {% else %} + missing + {% endif %} + + {% if source.enabled %} + enabled + {% else %} + disabled + {% endif %} + {{ source.created_at | dt }}
+
+ {% else %} +
+
No sources yet
+

+ {% if role == 'admin' %} + Add a Confluence instance above. Nothing is scanned until a source exists and a scan is launched. + {% else %} + An administrator configures the systems this deployment scans. + {% endif %} +

+
+ {% endif %} +
+ +{% if next_cursor %} +
+ Showing one page + + Next page + + +
+{% endif %} +{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templates/suppressions/list.html b/apps/api/src/iceberg_api/web/templates/suppressions/list.html new file mode 100644 index 0000000..b4b0a60 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templates/suppressions/list.html @@ -0,0 +1,172 @@ +{% extends "base.html" %} +{% block title %}Suppressions · {{ app_name }}{% endblock %} +{% block crumb %}Suppressions{% endblock %} + +{% block content %} +
+
+
Configuration
+

Suppressions

+
+
+
+ +

+ Tuning lives in data, not in the rule pack — a suppression takes effect immediately and without a + release. Creating one hides the findings it already covers; deleting it gives them back, with an + event saying why they returned. +

+ +{% if error %} + +{% endif %} + +{% if role in ('analyst', 'admin') %} +
+ + +
+
+ +
+ + +
+ + + +
+ +
+
+
+
+{% endif %} + +
+ + + + + Clear +
+ +
+ {% if suppressions %} +
+ + + + {% if role in ('analyst', 'admin') %}{% endif %} + + + {% for suppression in suppressions %} + + + + + + + + {% if role in ('analyst', 'admin') %} + + {% endif %} + + {% endfor %} + +
ScopePatternSourceReasonExpiresCreatedAction
{{ suppression.scope.value | humanize }}{{ suppression.pattern }} + {% if suppression.source_id %} + {{ source_names.get(suppression.source_id, suppression.source_id | short) }} + {% else %} + global + {% endif %} + {{ suppression.reason }}{{ suppression.expires_at | dt if suppression.expires_at else 'never' }}{{ suppression.created_at | dt }} + + + + + +
+
+ {% else %} +
+
Nothing is suppressed
+

Every finding this deployment records is in the active view.

+
+ {% endif %} +
+ +{% if next_cursor %} +
+ Showing one page + + Next page + + +
+{% endif %} +{% endblock %} diff --git a/apps/api/src/iceberg_api/web/templating.py b/apps/api/src/iceberg_api/web/templating.py new file mode 100644 index 0000000..9896568 --- /dev/null +++ b/apps/api/src/iceberg_api/web/templating.py @@ -0,0 +1,251 @@ +"""The Jinja environment, and the two ways a web route answers (#54). + +**The partial convention.** A route that a browser navigates to renders a full +page; a route HTMX calls renders a fragment. Which one is not guessed from +headers — it is the route's decision, expressed by calling :func:`render` (a page +template that extends ``base.html``) or :func:`render_partial` (a template under +``templates/partials/``, which extends nothing). Sniffing ``HX-Request`` to +decide would mean every template had two shapes and every bug had two places to +hide; a fragment URL that returns the same fragment to curl, to HTMX, and to a +browser's address bar is one you can debug. + +**The conventions a fragment follows.** + +* Fragments live in ``partials/`` and are named ``_thing.html``. +* A fragment owns one element with a stable id; the route that mutates it swaps + it whole (``hx-target="#thing" hx-swap="outerHTML"``). +* A mutation answers with the fragment its change affected, so the browser never + re-requests to find out what happened. +* Where the change is bigger than one region, the mutation answers + :func:`hx_redirect` and the browser does a normal navigation. + +Autoescaping is on for every template, which is what makes it safe to render a +redacted snippet or a Confluence page title — strings that came out of somebody +else's system and are the reason this UI is a stored-XSS target at all. +""" + +import uuid +from datetime import UTC, datetime +from typing import Any + +from fastapi import Request, Response +from fastapi.responses import HTMLResponse +from jinja2 import Environment, FileSystemLoader, StrictUndefined, select_autoescape + +from iceberg_api.web.assets import TEMPLATES_DIR, asset_manifest + +APP_NAME = "IcebergSST" + + +def _format_datetime(value: datetime | None) -> str: + """An absolute UTC timestamp. Never a local one. + + Every timestamp in this system is UTC, engines and analysts are in different + places, and a triage decision recorded at "14:02" is only useful if everyone + reading it means the same 14:02. + """ + if value is None: + return "—" + aware = value if value.tzinfo else value.replace(tzinfo=UTC) + return aware.astimezone(UTC).strftime("%Y-%m-%d %H:%M UTC") + + +def _format_ago(value: datetime | None) -> str: + """Coarse relative time, for "is this engine still alive?" at a glance.""" + if value is None: + return "never" + aware = value if value.tzinfo else value.replace(tzinfo=UTC) + seconds = (datetime.now(UTC) - aware).total_seconds() + if seconds < 0: + # A clock skew between the API and the reader, or a scheduled future time. + return "in the future" + for limit, divisor, unit in ( + (60, 1, "s"), + (3600, 60, "m"), + (86400, 3600, "h"), + ): + if seconds < limit: + return f"{int(seconds // divisor)}{unit} ago" + return f"{int(seconds // 86400)}d ago" + + +def _shorten(value: object, length: int = 12) -> str: + """A uuid's leading characters — enough to recognise, short enough to scan.""" + text = str(value) + return text if len(text) <= length else text[:length] + "…" + + +def _humanize(value: object) -> str: + """``accepted_risk`` → ``Accepted risk``. Enum values are wire format, not copy.""" + return str(value).replace("_", " ").capitalize() + + +def build_environment() -> Environment: + """The template environment. + + ``StrictUndefined`` turns a typo'd variable into a loud failure at render time + rather than a silently empty cell — in a findings table, a blank where a + severity should be is indistinguishable from a finding with no severity. + """ + env = Environment( + loader=FileSystemLoader(TEMPLATES_DIR), + autoescape=select_autoescape(default_for_string=True, default=True), + undefined=StrictUndefined, + trim_blocks=True, + lstrip_blocks=True, + auto_reload=False, + ) + env.filters["dt"] = _format_datetime + env.filters["ago"] = _format_ago + env.filters["short"] = _shorten + env.filters["humanize"] = _humanize + env.globals["app_name"] = APP_NAME + return env + + +_environment = build_environment() + + +def base_context(request: Request, **extra: Any) -> dict[str, Any]: + """What every template gets, whether it is a page or a fragment. + + ``csrf_token`` is here rather than passed per route because every mutating + form and every ``hx-`` request needs it, and a token supplied by hand is one + somebody eventually forgets. + """ + context: dict[str, Any] = { + "request": request, + "path": request.url.path, + "assets": asset_manifest(), + "user": None, + "role": None, + "csrf_token": "", + } + context.update(extra) + return context + + +def render( + request: Request, + template: str, + context: dict[str, Any] | None = None, + *, + status_code: int = 200, + headers: dict[str, str] | None = None, +) -> HTMLResponse: + """Render a full page (a template that extends ``base.html``).""" + body = _environment.get_template(template).render(base_context(request, **(context or {}))) + return HTMLResponse(body, status_code=status_code, headers=headers) + + +def render_partial( + request: Request, + template: str, + context: dict[str, Any] | None = None, + *, + status_code: int = 200, + headers: dict[str, str] | None = None, +) -> HTMLResponse: + """Render a fragment from ``partials/`` — no layout, no ````.""" + return render( + request, + f"partials/{template}", + context, + status_code=status_code, + headers=headers, + ) + + +def viewer_context(viewer: Any) -> dict[str, Any]: + """Identity for the shell: who is signed in, their role, and their token. + + Takes the :class:`~iceberg_api.web.dependencies.Viewer` structurally rather + than by import, so the template layer stays free of the auth layer. + """ + return { + "user": viewer.user, + "role": viewer.user.role.value, + "csrf_token": viewer.csrf_token, + } + + +def render_page( + request: Request, + template: str, + viewer: Any, + context: dict[str, Any] | None = None, + *, + status_code: int = 200, + headers: dict[str, str] | None = None, +) -> HTMLResponse: + """Render a page inside the authenticated shell.""" + return render( + request, + template, + viewer_context(viewer) | (context or {}), + status_code=status_code, + headers=headers, + ) + + +def render_fragment( + request: Request, + template: str, + viewer: Any, + context: dict[str, Any] | None = None, + *, + status_code: int = 200, + headers: dict[str, str] | None = None, +) -> HTMLResponse: + """Render a fragment with the same identity a page would have. + + A fragment needs ``user``, ``role``, and ``csrf_token`` for exactly the + reasons a page does: it contains forms that must carry a token and controls + that only some roles are offered. Without this, a partial rendered by a + mutation would differ from the identical markup rendered inside its page. + """ + return render_partial( + request, + template, + viewer_context(viewer) | (context or {}), + status_code=status_code, + headers=headers, + ) + + +def hx_redirect(url: str) -> Response: + """Tell HTMX to navigate rather than swap. + + A 302 would be followed by ``fetch`` and the login page (or the new detail + page) would be swapped into whatever element issued the request. ``HX-Redirect`` + is the header that makes the browser actually go there. + """ + return Response(status_code=204, headers={"HX-Redirect": url}) + + +def hx_refresh() -> Response: + """Tell HTMX to reload the current page — for a change with no single target.""" + return Response(status_code=204, headers={"HX-Refresh": "true"}) + + +def is_htmx(request: Request) -> bool: + """Whether HTMX issued this request. + + Used only to choose the *failure* shape — a redirect header instead of a + redirect response — never to choose which template renders. + """ + return request.headers.get("HX-Request") == "true" + + +def id_or_none(value: str | None) -> uuid.UUID | None: + """Parse an optional uuid from a query string, treating junk as absent. + + A filter arriving as a non-uuid is a stale bookmark or a hand-edited URL. The + honest answer is the unfiltered list, not a 422 on a read-only page. + """ + if not value: + return None + try: + return uuid.UUID(value) + except ValueError: + return None diff --git a/apps/api/tests/test_notifications.py b/apps/api/tests/test_notifications.py new file mode 100644 index 0000000..0d9d92a --- /dev/null +++ b/apps/api/tests/test_notifications.py @@ -0,0 +1,361 @@ +"""Notification-channel CRUD (#72). + +A webhook channel is a deliberate egress path — it carries redacted snippets and +resource locations off the deployment — so these tests care about three things +above all: only an admin may configure one, the secret that signs its payloads is +sealed and never returned, and every change is in the audit trail with the +destination it went to. +""" + +import uuid +from collections.abc import Callable +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from iceberg_core.enums import NotificationChannelType, UserRole +from iceberg_core.models import AUDIT_TARGET_CHANNEL, AuditEvent, NotificationChannel, User +from sqlmodel import Session, select + +WEBHOOK = { + "name": "security-alerts", + "type": "webhook", + "config": {"url": "https://chat.example.com/hooks/abc"}, + "secret": "payload-signing-secret", + "event_filter": {"min_severity": "high"}, +} + + +def _create( + client: TestClient, api: str, headers: dict[str, str], **overrides: object +) -> dict[str, Any]: + response = client.post( + f"{api}/notifications/channels", json=WEBHOOK | overrides, headers=headers + ) + assert response.status_code == 201, response.text + created: dict[str, Any] = response.json() + return created + + +def test_an_admin_can_create_a_webhook_channel( + client: TestClient, + api: str, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + body = _create(client, api, headers) + + assert body["name"] == "security-alerts" + assert body["type"] == NotificationChannelType.WEBHOOK.value + assert body["config"]["url"] == "https://chat.example.com/hooks/abc" + assert body["event_filter"]["min_severity"] == "high" + assert body["has_secret"] is True + + channel = session.exec(select(NotificationChannel)).one() + assert channel.config["secret_ref"] + assert "payload-signing-secret" not in str(channel.config) + + +def test_no_response_carries_the_secret_or_its_sealed_ref( + client: TestClient, + api: str, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """A ref plus a leaked master key is a credential; no client needs one.""" + headers = login_as(make_user(UserRole.ADMIN)) + created = _create(client, api, headers) + ref = session.exec(select(NotificationChannel)).one().config["secret_ref"] + + listed = client.get(f"{api}/notifications/channels", headers=headers) + read = client.get(f"{api}/notifications/channels/{created['id']}", headers=headers) + + for response in (listed, read): + assert "payload-signing-secret" not in response.text + assert ref not in response.text + assert "secret_ref" not in response.text + + +@pytest.mark.parametrize("role", [UserRole.VIEWER, UserRole.ANALYST]) +def test_only_an_admin_may_read_or_write_channels( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], + role: UserRole, +) -> None: + """Configuring an egress path is an operator action (ADR 0005).""" + headers = login_as(make_user(role)) + + assert client.get(f"{api}/notifications/channels", headers=headers).status_code == 403 + assert ( + client.post(f"{api}/notifications/channels", json=WEBHOOK, headers=headers).status_code + == 403 + ) + + +def test_an_unauthenticated_caller_gets_401_not_403(client: TestClient, api: str) -> None: + assert client.get(f"{api}/notifications/channels").status_code == 401 + + +def test_a_mutation_without_a_csrf_token_is_refused( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + login_as(make_user(UserRole.ADMIN)) + + response = client.post(f"{api}/notifications/channels", json=WEBHOOK) + + assert response.status_code == 403 + + +def test_a_webhook_url_must_be_an_absolute_http_url( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """A bad URL fails when the channel is saved, not at 3am on a critical finding.""" + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + f"{api}/notifications/channels", + json=WEBHOOK | {"config": {"url": "chat.example.com/hooks/abc"}}, + headers=headers, + ) + + assert response.status_code == 422 + assert "http://" in response.text + + +def test_a_credential_carrying_header_is_refused_with_a_pointer_at_secret( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """Headers are stored as plain JSON. A token in one is a plaintext secret at + rest — the exact thing this product exists to find in other systems.""" + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + f"{api}/notifications/channels", + json=WEBHOOK + | { + "config": { + "url": "https://chat.example.com/hooks/abc", + "headers": {"Authorization": "Bearer hunter2"}, + } + }, + headers=headers, + ) + + assert response.status_code == 422 + assert "secret" in response.text + + +def test_a_header_value_cannot_smuggle_a_line_break( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + f"{api}/notifications/channels", + json=WEBHOOK + | { + "config": { + "url": "https://chat.example.com/hooks/abc", + "headers": {"X-Route": "a\r\nX-Injected: b"}, + } + }, + headers=headers, + ) + + assert response.status_code == 422 + + +def test_a_caller_supplied_secret_ref_is_dropped( + client: TestClient, + api: str, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """The ref is minted here. Accepting one would let a caller point a channel + at another object's sealed material.""" + headers = login_as(make_user(UserRole.ADMIN)) + + _create( + client, + api, + headers, + config={"url": "https://chat.example.com/hooks/abc", "secret_ref": "somebody-elses-ref"}, + secret=None, + ) + + channel = session.exec(select(NotificationChannel)).one() + assert "secret_ref" not in channel.config + + +def test_an_email_channel_validates_its_recipients( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + good = client.post( + f"{api}/notifications/channels", + json={ + "name": "secops-mail", + "type": "email", + "config": {"recipients": ["secops@example.com"]}, + }, + headers=headers, + ) + bad = client.post( + f"{api}/notifications/channels", + json={"name": "broken", "type": "email", "config": {"recipients": ["not-an-address"]}}, + headers=headers, + ) + + assert good.status_code == 201 + assert bad.status_code == 422 + + +def test_a_duplicate_name_is_a_conflict( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + _create(client, api, headers) + + response = client.post(f"{api}/notifications/channels", json=WEBHOOK, headers=headers) + + assert response.status_code == 409 + + +def test_editing_the_config_keeps_the_sealed_secret( + client: TestClient, + api: str, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """Fixing a typo in a URL is not a request to drop the signing key.""" + headers = login_as(make_user(UserRole.ADMIN)) + created = _create(client, api, headers) + original_ref = session.exec(select(NotificationChannel)).one().config["secret_ref"] + + response = client.patch( + f"{api}/notifications/channels/{created['id']}", + json={"config": {"url": "https://chat.example.com/hooks/def"}}, + headers=headers, + ) + + assert response.status_code == 200 + assert response.json()["has_secret"] is True + channel = session.exec(select(NotificationChannel)).one() + assert channel.config["url"].endswith("/def") + assert channel.config["secret_ref"] == original_ref + + +def test_supplying_a_secret_rotates_it_and_records_a_separate_event( + client: TestClient, + api: str, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + created = _create(client, api, headers) + original_ref = session.exec(select(NotificationChannel)).one().config["secret_ref"] + + response = client.patch( + f"{api}/notifications/channels/{created['id']}", + json={"secret": "a-rotated-secret"}, + headers=headers, + ) + + assert response.status_code == 200 + channel = session.exec(select(NotificationChannel)).one() + assert channel.config["secret_ref"] != original_ref + + actions = [ + event.action + for event in session.exec( + select(AuditEvent).where(AuditEvent.target_type == AUDIT_TARGET_CHANNEL) + ) + ] + # "who changed this channel" and "who replaced its key" are different questions. + assert actions.count("channel.secret_set") == 2 + + +def test_a_channel_can_be_disabled_without_being_deleted( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + created = _create(client, api, headers) + + response = client.patch( + f"{api}/notifications/channels/{created['id']}", json={"enabled": False}, headers=headers + ) + + assert response.status_code == 200 + assert response.json()["enabled"] is False + + +def test_creating_and_deleting_records_the_destination_never_the_secret( + client: TestClient, + api: str, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """ "Who pointed our findings at that URL" must always have an answer.""" + headers = login_as(make_user(UserRole.ADMIN)) + created = _create(client, api, headers) + + assert ( + client.delete(f"{api}/notifications/channels/{created['id']}", headers=headers).status_code + == 204 + ) + + events = list( + session.exec(select(AuditEvent).where(AuditEvent.target_type == AUDIT_TARGET_CHANNEL)) + ) + actions = {event.action for event in events} + assert {"channel.created", "channel.secret_set", "channel.deleted"} == actions + destinations = { + event.detail.get("destination") for event in events if "destination" in event.detail + } + assert destinations == {"https://chat.example.com/hooks/abc"} + assert all("payload-signing-secret" not in str(event.detail) for event in events) + + +def test_reading_an_unknown_channel_is_a_404( + client: TestClient, + api: str, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.get(f"{api}/notifications/channels/{uuid.uuid4()}", headers=headers) + + assert response.status_code == 404 diff --git a/apps/api/tests/test_web_invariants.py b/apps/api/tests/test_web_invariants.py new file mode 100644 index 0000000..96be919 --- /dev/null +++ b/apps/api/tests/test_web_invariants.py @@ -0,0 +1,255 @@ +"""Invariants that keep the web UI a client of the API rather than a second one. + +The M3 UI could very easily have grown its own queries — it renders lists and +details, and a `select()` is right there. That would give the product two +implementations of "which findings may this person see", and the one nobody +audits is the one that leaks. So the boundary is asserted here rather than +described in a comment: + +* no module under ``iceberg_api.web`` touches the database; +* every mutating browser route carries CSRF protection; +* no HTML endpoint appears in the OpenAPI document; +* the vendored assets on disk are the ones the lock file describes. +""" + +import ast +import base64 +import hashlib +import json +import re +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.routing import APIRoute +from iceberg_api.app import API_PREFIX +from iceberg_api.auth.dependencies import require_csrf +from iceberg_api.web import assets as web_assets +from iceberg_api.web.routes import router as web_router + +WEB_PACKAGE = Path(web_assets.__file__).resolve().parent +WEB_MODULES = sorted(WEB_PACKAGE.rglob("*.py")) + +#: Ways a module would reach the database. The web layer must use none of them — +#: it holds a `Session` only to hand it to the API handler it is calling. +FORBIDDEN_CALLS = frozenset({"select", "exec", "commit", "refresh", "flush"}) +FORBIDDEN_IMPORTS = ("sqlmodel", "sqlalchemy", "iceberg_core.db") + +#: The shared Iceberg palette, copied from the canonical source — +#: ``IcebergCM/src/icebergcm/web/static/app.css``, which carries the same values +#: as ``iceberg``'s ``iceberg.css`` (docs/web.md § Provenance). Pinned here rather +#: than read from a sibling checkout because CI has no sibling checkout: the point +#: is to fail when *this* repo drifts, and the fix is to go back to that file. +CANONICAL_TOKENS = { + "--accent": "oklch(0.66 0.118 226)", + "--ink": "oklch(0.26 0.026 262)", + "--paper": "oklch(0.984 0.006 240)", + "--rail": "oklch(0.225 0.024 256)", + "--line": "oklch(0.915 0.008 248)", +} + + +def test_the_web_package_has_modules_to_check() -> None: + """A guard on the guards: an empty glob would pass every test below.""" + assert len(WEB_MODULES) >= 10 + + +@pytest.mark.parametrize("module", WEB_MODULES, ids=lambda path: path.name) +def test_no_web_module_imports_the_orm(module: Path) -> None: + source = module.read_text(encoding="utf-8") + offenders = [name for name in FORBIDDEN_IMPORTS if f"import {name}" in source] + + assert offenders == [], ( + f"{module.name} imports {offenders}; the web layer reaches the database " + "only by calling an API route handler" + ) + + +@pytest.mark.parametrize("module", WEB_MODULES, ids=lambda path: path.name) +def test_no_web_module_builds_a_query_or_commits(module: Path) -> None: + """`db` is passed through to an API handler, never used here.""" + tree = ast.parse(module.read_text(encoding="utf-8"), filename=str(module)) + offenders: list[str] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + target = node.func + name = ( + target.attr + if isinstance(target, ast.Attribute) + else target.id + if isinstance(target, ast.Name) + else None + ) + if name in FORBIDDEN_CALLS: + offenders.append(f"{name}() at line {node.lineno}") + + assert offenders == [], f"{module.name} touches the database directly: {offenders}" + + +def _routes_of(router: object) -> list[APIRoute]: + """Flatten a router tree into its API routes. + + Walked from the web router itself rather than from ``app.routes``: FastAPI + keeps an included router as a lazy placeholder, and the paths inside it are + the *unprefixed* ones — so a scan of the application would not be able to + tell ``/findings`` the page from ``/findings`` the API route that is mounted + under ``/api/v1``. Starting at the web router removes the ambiguity. + """ + from fastapi.routing import _IncludedRouter + + found: list[APIRoute] = [] + pending = list(getattr(router, "routes", [])) + while pending: + route = pending.pop() + if isinstance(route, _IncludedRouter): + pending.extend(route.original_router.routes) + elif isinstance(route, APIRoute): + found.append(route) + return found + + +WEB_ROUTES = _routes_of(web_router) + + +def test_the_web_router_carries_every_screen() -> None: + paths = {route.path for route in WEB_ROUTES} + + assert {"/", "/login", "/logout", "/findings/{finding_id}", "/scans/{scan_id}/status"} <= paths + assert len(paths) >= 15 + + +def test_no_html_endpoint_appears_in_the_openapi_document(app: FastAPI) -> None: + """The schema is the API's contract; HTML routes in it would describe a second.""" + documented = set(app.openapi()["paths"]) + + assert all(route.path not in documented for route in WEB_ROUTES) + assert all(not route.include_in_schema for route in WEB_ROUTES) + # And the API's own surface is still documented, so this is not passing by + # having accidentally emptied the schema. + assert f"{API_PREFIX}/findings" in documented + + +def test_every_mutating_web_route_is_csrf_protected() -> None: + """Cookie auth means a browser attaches credentials to cross-site requests.""" + unprotected: list[str] = [] + for route in WEB_ROUTES: + mutating = (route.methods or set()) & {"POST", "PUT", "PATCH", "DELETE"} + if not mutating: + continue + if not _depends_on(route, require_csrf): + unprotected.append(f"{sorted(mutating)} {route.path}") + + assert unprotected == [], f"mutating web routes without CSRF protection: {unprotected}" + + +def _depends_on(route: APIRoute, target: object) -> bool: + """Whether ``target`` appears anywhere in a route's dependency tree.""" + pending = [route.dependant] + while pending: + dependant = pending.pop() + if dependant.call is target: + return True + pending.extend(dependant.dependencies) + return False + + +# ─── Vendored assets ───────────────────────────────────────────────────────── + + +def test_the_static_tree_is_served_from_this_origin(app: FastAPI) -> None: + """`script-src 'self'` is only satisfiable if the app actually serves them.""" + from fastapi.testclient import TestClient + + manifest = json.loads(web_assets.LOCK_PATH.read_text(encoding="utf-8")) + with TestClient(app) as client: + for entry in manifest.values(): + response = client.get(f"/static/{entry['path']}") + assert response.status_code == 200, entry["path"] + + assert client.get("/static/css/iceberg.css").status_code == 200 + assert client.get("/static/js/tags.js").status_code == 200 + assert client.get("/static/img/icebergai-mark.svg").status_code == 200 + + +def test_every_vendored_asset_matches_its_recorded_integrity() -> None: + """CI verifies the committed files offline, without a network call. + + A hand-edited vendor file, a truncated download, or a lock that was not + regenerated all produce the same symptom in a browser — the asset is blocked + by SRI and the page silently loses its behaviour. This catches it here. + """ + manifest = json.loads(web_assets.LOCK_PATH.read_text(encoding="utf-8")) + + for name, entry in sorted(manifest.items()): + path = web_assets.STATIC_DIR / entry["path"] + assert path.is_file(), f"{name}: {entry['path']} is missing" + digest = hashlib.sha384(path.read_bytes()).digest() + expected = "sha384-" + base64.b64encode(digest).decode("ascii") + assert expected == entry["integrity"], ( + f"{name} does not match its lock entry; re-run web/vendor_assets.py" + ) + + +def test_every_declared_font_file_is_present() -> None: + manifest = json.loads(web_assets.LOCK_PATH.read_text(encoding="utf-8")) + missing = [ + filename + for filename in manifest["fonts"]["files"] + if not (web_assets.STATIC_DIR / "fonts" / filename).is_file() + ] + + assert missing == [], f"declared but absent: {missing}" + + +def test_the_font_stylesheet_points_only_at_this_origin() -> None: + """A Google Fonts URL would need `font-src` to name a third party, and would + break the app in an air-gapped deployment.""" + css = (web_assets.STATIC_DIR / "css" / "vendor" / "fonts.css").read_text(encoding="utf-8") + + assert "fonts.gstatic.com" not in css + assert "fonts.googleapis.com" not in css + assert "/static/fonts/" in css + + +def test_the_alpine_bundle_is_the_csp_build() -> None: + """The standard build uses `new Function()` and cannot run under this policy.""" + bundle = (web_assets.STATIC_DIR / "js" / "vendor" / "alpine.min.js").read_text( + encoding="utf-8", errors="replace" + ) + + # The CSP build ships its own expression interpreter and never constructs a + # function from a string; the standard build's evaluator is `new Function`. + assert "new Function(" not in bundle + + +# ─── Design system ─────────────────────────────────────────────────────────── + + +def test_the_stylesheet_carries_the_shared_iceberg_tokens() -> None: + """Same palette as iceberg and IcebergCM — one visual language, three apps.""" + css = (web_assets.STATIC_DIR / "css" / "iceberg.css").read_text(encoding="utf-8") + + for token, value in CANONICAL_TOKENS.items(): + assert re.search(rf"{re.escape(token)}:\s*{re.escape(value)}", css), ( + f"{token} has drifted from the shared palette" + ) + + +def test_the_stylesheet_has_a_dark_variant() -> None: + css = (web_assets.STATIC_DIR / "css" / "iceberg.css").read_text(encoding="utf-8") + + assert "@media (prefers-color-scheme: dark)" in css + assert "color-scheme: dark" in css + + +def test_the_stylesheet_hard_codes_no_hex_colours_outside_the_rail() -> None: + """Colours come from tokens so light↔dark switches for free. + + `#fff` on the dark rail is the deliberate exception: that chrome is dark in + both themes, so its foreground is a constant rather than a token. + """ + css = (web_assets.STATIC_DIR / "css" / "iceberg.css").read_text(encoding="utf-8") + hexes = set(re.findall(r"#[0-9a-fA-F]{3,8}\b", css)) + + assert hexes <= {"#fff"}, f"hard-coded colours found: {sorted(hexes)}" diff --git a/apps/api/tests/test_web_screens.py b/apps/api/tests/test_web_screens.py new file mode 100644 index 0000000..8d0933f --- /dev/null +++ b/apps/api/tests/test_web_screens.py @@ -0,0 +1,943 @@ +"""The operational screens (#55, #56, #57, #58, #72). + +Each test asserts the epic's acceptance criterion the way a user would meet it: +the screen renders, the action drives the API route it is supposed to drive, the +change lands in the database, and the things that must never appear on a page — +a credential, a webhook secret, anything derived from a detected secret — do not. +""" + +import uuid +from collections.abc import Callable + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from iceberg_api.sources.routes import get_prober +from iceberg_api.sources.schemas import ConnectivityResult +from iceberg_core.enums import ( + FindingEventKind, + FindingState, + NotificationChannelType, + ScanStatus, + ScanTaskKind, + ScanTaskStatus, + ScanTrigger, + Severity, + SourceType, + UserRole, +) +from iceberg_core.models import ( + AUDIT_TARGET_CHANNEL, + AUDIT_TARGET_USER, + Engine, + Finding, + FindingEvent, + NotificationChannel, + Scan, + ScanTask, + Source, + Suppression, + User, +) +from pydantic import SecretStr +from sqlmodel import Session, select + +CONFLUENCE_FORM = { + "name": "confluence-eng", + "type": "confluence", + "base_url": "https://example.atlassian.net/wiki", + "email": "scanner@example.com", + "api_prefix": "", + "credential": "super-secret-api-token", + "include_comments": "on", + "include_attachments": "on", + "enabled": "on", +} + + +# ─── #55 Sources ───────────────────────────────────────────────────────────── + + +def test_creating_a_source_from_the_form_drives_the_sources_api( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + "/sources", + data=CONFLUENCE_FORM | {"csrf_token": headers["X-CSRF-Token"], "spaces": ["ENG", "OPS"]}, + headers=headers, + ) + + assert response.status_code == 204 + assert response.headers["HX-Redirect"].startswith("/sources/") + + source = session.exec(select(Source).where(Source.name == "confluence-eng")).one() + assert source.type is SourceType.CONFLUENCE + assert source.connection["base_url"] == "https://example.atlassian.net/wiki" + assert source.connection["spaces"] == ["ENG", "OPS"] + assert source.connection["email"] == "scanner@example.com" + # Sealed through the secret store, not stored as typed. + assert source.credential_ref is not None + assert "super-secret-api-token" not in str(source.credential_ref) + + +def test_the_source_form_never_echoes_the_credential_back( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """#55's acceptance criterion: credential input is never echoed back.""" + headers = login_as(make_user(UserRole.ADMIN)) + client.post( + "/sources", + data=CONFLUENCE_FORM | {"csrf_token": headers["X-CSRF-Token"]}, + headers=headers, + ) + source = session.exec(select(Source).where(Source.name == "confluence-eng")).one() + + list_page = client.get("/sources").text + detail_page = client.get(f"/sources/{source.id}").text + + for page in (list_page, detail_page): + assert "super-secret-api-token" not in page + # Not the sealed ref either: a ref plus a leaked master key is a credential. + assert str(source.credential_ref) not in page + assert "stored" in detail_page + + +def test_a_rejected_source_re_renders_the_form_with_the_reason( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + """A bad base_url fails on save, and the analyst keeps what they typed.""" + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + "/sources", + data=CONFLUENCE_FORM | {"csrf_token": headers["X-CSRF-Token"], "base_url": "not-a-url"}, + headers=headers, + ) + + assert response.status_code == 200 + assert "base_url must start with http:// or https://" in response.text + assert 'id="source-form"' in response.text + + +def test_a_viewer_cannot_create_a_source_from_the_ui( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + headers = login_as(make_user(UserRole.VIEWER)) + + response = client.post( + "/sources", data=CONFLUENCE_FORM | {"csrf_token": headers["X-CSRF-Token"]}, headers=headers + ) + + assert response.status_code == 403 + + +def test_editing_a_source_without_a_credential_keeps_the_stored_one( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + client.post( + "/sources", + data=CONFLUENCE_FORM | {"csrf_token": headers["X-CSRF-Token"]}, + headers=headers, + ) + source = session.exec(select(Source).where(Source.name == "confluence-eng")).one() + original_ref = source.credential_ref + + response = client.post( + f"/sources/{source.id}", + data={ + "csrf_token": headers["X-CSRF-Token"], + "name": "confluence-eng-renamed", + "base_url": "https://example.atlassian.net/wiki", + "email": "scanner@example.com", + "api_prefix": "", + "credential": "", + "enabled": "on", + }, + headers=headers, + ) + + assert response.status_code == 204 + session.refresh(source) + assert source.name == "confluence-eng-renamed" + assert source.credential_ref == original_ref + + +def test_the_connectivity_check_reports_what_the_probe_said( + client: TestClient, + app: FastAPI, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + client.post( + "/sources", + data=CONFLUENCE_FORM | {"csrf_token": headers["X-CSRF-Token"]}, + headers=headers, + ) + source = session.exec(select(Source).where(Source.name == "confluence-eng")).one() + + async def _reachable(_source: Source, _credential: SecretStr) -> ConnectivityResult: + return ConnectivityResult(reachable=True, status_code=200, detail="answered as scanner") + + app.dependency_overrides[get_prober] = lambda: _reachable + + response = client.post(f"/sources/{source.id}/test", headers=headers) + + assert response.status_code == 200 + assert "Reachable" in response.text + assert "answered as scanner" in response.text + assert "super-secret-api-token" not in response.text + + +def test_testing_a_source_with_no_credential_says_so_rather_than_500ing( + client: TestClient, + make_source: Callable[..., Source], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + source = make_source() + + response = client.post(f"/sources/{source.id}/test", headers=headers) + + assert response.status_code == 200 + assert "Could not test" in response.text + assert "no stored credential" in response.text + + +# ─── #56 Scans ─────────────────────────────────────────────────────────────── + + +@pytest.fixture(name="scan_with_tasks") +def scan_with_tasks_fixture( + session: Session, make_source: Callable[..., Source] +) -> Callable[..., Scan]: + """A scan plus two tasks, so the progress bar has something to count.""" + + def factory(status: ScanStatus = ScanStatus.RUNNING, finished: int = 1) -> Scan: + source = make_source() + scan = Scan(source_id=source.id, trigger=ScanTrigger.MANUAL, status=status) + session.add(scan) + session.commit() + for index in range(2): + session.add( + ScanTask( + scan_id=scan.id, + kind=ScanTaskKind.FETCH, + status=( + ScanTaskStatus.COMPLETED if index < finished else ScanTaskStatus.RUNNING + ), + spec={"page_id": f"page-{index}"}, + ) + ) + session.commit() + return scan + + return factory + + +def test_launching_a_scan_drives_the_sources_scan_route( + client: TestClient, + session: Session, + make_source: Callable[..., Source], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], + dispatcher: object, +) -> None: + """#56's acceptance criterion: launching drives POST /sources/{id}/scan.""" + headers = login_as(make_user(UserRole.ANALYST)) + source = make_source() + + response = client.post(f"/sources/{source.id}/scan", headers=headers) + + assert response.status_code == 204 + scan = session.exec(select(Scan).where(Scan.source_id == source.id)).one() + assert response.headers["HX-Redirect"] == f"/scans/{scan.id}" + # One scan, one discovery task, one dispatch (ADR 0009). + assert len(dispatcher.enqueued) == 1 # type: ignore[attr-defined] + + +def test_a_viewer_cannot_launch_a_scan( + client: TestClient, + make_source: Callable[..., Source], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.VIEWER)) + + response = client.post(f"/sources/{make_source().id}/scan", headers=headers) + + assert response.status_code == 403 + + +def test_a_running_scan_polls_and_shows_task_progress( + client: TestClient, + scan_with_tasks: Callable[..., Scan], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + login_as(make_user(UserRole.VIEWER)) + scan = scan_with_tasks(ScanStatus.RUNNING, finished=1) + + response = client.get(f"/scans/{scan.id}") + + assert response.status_code == 200 + assert f'hx-get="/scans/{scan.id}/status"' in response.text + assert 'hx-trigger="every 3s"' in response.text + assert 'aria-valuenow="50"' in response.text + assert "1 of 2 task(s) finished" in response.text + + +def test_a_finished_scan_stops_polling( + client: TestClient, + scan_with_tasks: Callable[..., Scan], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """The terminal fragment carries no trigger, so the browser stops asking.""" + login_as(make_user(UserRole.VIEWER)) + scan = scan_with_tasks(ScanStatus.COMPLETED, finished=2) + + response = client.get(f"/scans/{scan.id}/status") + + assert response.status_code == 200 + assert "hx-trigger" not in response.text + assert 'aria-valuenow="100"' in response.text + + +def test_cancelling_a_scan_drives_the_cancel_route( + client: TestClient, + session: Session, + scan_with_tasks: Callable[..., Scan], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ANALYST)) + scan = scan_with_tasks(ScanStatus.RUNNING, finished=0) + + response = client.post(f"/scans/{scan.id}/cancel", headers=headers) + + assert response.status_code == 200 + session.refresh(scan) + assert scan.status is ScanStatus.CANCELLED + assert "cancelled" in response.text + + +def test_cancelling_a_finished_scan_says_so_instead_of_erroring( + client: TestClient, + scan_with_tasks: Callable[..., Scan], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ANALYST)) + scan = scan_with_tasks(ScanStatus.COMPLETED, finished=2) + + response = client.post(f"/scans/{scan.id}/cancel", headers=headers) + + assert response.status_code == 200 + assert "Nothing to cancel" in response.text + + +# ─── #57 Findings ──────────────────────────────────────────────────────────── + + +def test_the_findings_table_applies_only_the_filters_in_the_url( + client: TestClient, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + login_as(make_user(UserRole.VIEWER)) + critical = make_finding(severity=Severity.CRITICAL, rule_id="aws-access-key") + low = make_finding(severity=Severity.LOW, rule_id="generic-password") + + filtered = client.get("/findings?severity=critical").text + unfiltered = client.get("/findings").text + + assert str(critical.id) in filtered + assert str(low.id) not in filtered + assert str(critical.id) in unfiltered and str(low.id) in unfiltered + + +def test_the_finding_detail_shows_the_redacted_snippet_and_no_hash( + client: TestClient, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """#57: no plaintext shown — and nothing derived from the secret either.""" + login_as(make_user(UserRole.VIEWER)) + finding = make_finding(redacted_snippet="AKIA****************WXYZ", secret_hash="a" * 64) + + response = client.get(f"/findings/{finding.id}") + + assert response.status_code == 200 + assert "AKIA****************WXYZ" in response.text + assert "a" * 64 not in response.text + + +def test_a_hostile_snippet_is_escaped_not_executed( + client: TestClient, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """The snippet came out of somebody else's Confluence. Autoescaping is the + reason this page is not a stored-XSS delivery mechanism.""" + login_as(make_user(UserRole.VIEWER)) + finding = make_finding( + redacted_snippet="", + resource_locator={"path": ""}, + ) + + body = client.get(f"/findings/{finding.id}").text + + assert "" not in body + assert "<img src=x onerror=alert(1)>" in body + assert "" not in body + + +def test_triage_updates_the_finding_and_renders_its_new_history( + client: TestClient, + session: Session, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """#57's acceptance criterion, end to end.""" + analyst = make_user(UserRole.ANALYST) + headers = login_as(analyst) + finding = make_finding() + + response = client.post( + f"/findings/{finding.id}/triage", + data={ + "csrf_token": headers["X-CSRF-Token"], + "state": FindingState.FALSE_POSITIVE.value, + "comment": "Example credential in the onboarding guide", + "assignee": "none", + }, + headers=headers, + ) + + assert response.status_code == 200 + session.refresh(finding) + assert finding.state is FindingState.FALSE_POSITIVE + + events = session.exec(select(FindingEvent).where(FindingEvent.finding_id == finding.id)).all() + kinds = {event.kind for event in events} + assert FindingEventKind.STATE_CHANGE in kinds + # The history is rendered with the change that produced it. + assert "Example credential in the onboarding guide" in response.text + assert analyst.display_name in response.text + assert "False positive" in response.text + + +def test_an_illegal_transition_is_refused_with_its_reason_and_changes_nothing( + client: TestClient, + session: Session, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """Judgement → judgement is a 409, and the assignee must not move either.""" + analyst = make_user(UserRole.ANALYST) + headers = login_as(analyst) + finding = make_finding(state=FindingState.FALSE_POSITIVE) + + response = client.post( + f"/findings/{finding.id}/triage", + data={ + "csrf_token": headers["X-CSRF-Token"], + "state": FindingState.ACCEPTED_RISK.value, + "assignee": str(analyst.id), + }, + headers=headers, + ) + + assert response.status_code == 200 + assert "Not applied" in response.text + session.refresh(finding) + assert finding.state is FindingState.FALSE_POSITIVE + assert finding.assignee_id is None + + +def test_a_viewer_sees_the_history_but_is_offered_no_triage_form( + client: TestClient, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + login_as(make_user(UserRole.VIEWER)) + finding = make_finding() + + body = client.get(f"/findings/{finding.id}").text + + assert "History" in body + assert 'hx-post="/findings/' not in body + assert "has to have a name attached to it" in body + + +def test_a_viewer_cannot_triage( + client: TestClient, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.VIEWER)) + finding = make_finding() + + response = client.post( + f"/findings/{finding.id}/triage", + data={"csrf_token": headers["X-CSRF-Token"], "state": FindingState.RESOLVED.value}, + headers=headers, + ) + + assert response.status_code == 403 + + +# ─── #58 Suppressions, schedules, engines ──────────────────────────────────── + + +def test_creating_a_suppression_hides_the_findings_it_already_covers( + client: TestClient, + session: Session, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ANALYST)) + finding = make_finding(rule_id="generic-password") + + response = client.post( + "/suppressions", + data={ + "csrf_token": headers["X-CSRF-Token"], + "scope": "rule", + "pattern": "generic-password", + "reason": "Placeholder values in the onboarding guide", + "source_id": "", + }, + headers=headers, + ) + + assert response.status_code == 204 + session.refresh(finding) + assert finding.suppressed_at is not None + assert session.exec(select(Suppression)).one().reason.startswith("Placeholder") + + +def test_a_suppression_without_a_reason_is_refused_with_the_reason_why( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + headers = login_as(make_user(UserRole.ANALYST)) + + response = client.post( + "/suppressions", + data={ + "csrf_token": headers["X-CSRF-Token"], + "scope": "rule", + "pattern": "generic-password", + "reason": " ", + }, + headers=headers, + ) + + assert response.status_code == 204 + assert "error=" in response.headers["HX-Redirect"] + + +def test_deleting_a_suppression_releases_what_it_was_hiding( + client: TestClient, + session: Session, + make_finding: Callable[..., Finding], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ANALYST)) + finding = make_finding(rule_id="generic-password") + client.post( + "/suppressions", + data={ + "csrf_token": headers["X-CSRF-Token"], + "scope": "rule", + "pattern": "generic-password", + "reason": "Placeholder values", + }, + headers=headers, + ) + suppression = session.exec(select(Suppression)).one() + + response = client.request("DELETE", f"/suppressions/{suppression.id}", headers=headers) + + assert response.status_code == 204 + session.refresh(finding) + assert finding.suppressed_at is None + + +def test_a_viewer_cannot_create_a_suppression( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + headers = login_as(make_user(UserRole.VIEWER)) + + response = client.post( + "/suppressions", + data={ + "csrf_token": headers["X-CSRF-Token"], + "scope": "rule", + "pattern": "x", + "reason": "y", + }, + headers=headers, + ) + + assert response.status_code == 403 + + +def test_creating_a_schedule_computes_its_next_run( + client: TestClient, + session: Session, + make_source: Callable[..., Source], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + source = make_source() + + response = client.post( + "/schedules", + data={ + "csrf_token": headers["X-CSRF-Token"], + "source_id": str(source.id), + "cron": "0 3 * * *", + "enabled": "on", + }, + headers=headers, + ) + + assert response.status_code == 204 + from iceberg_core.models import Schedule + + schedule = session.exec(select(Schedule)).one() + assert schedule.cron == "0 3 * * *" + assert schedule.next_run_at is not None + + +def test_an_invalid_cron_is_refused_with_an_explanation( + client: TestClient, + make_source: Callable[..., Source], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + "/schedules", + data={ + "csrf_token": headers["X-CSRF-Token"], + "source_id": str(make_source().id), + "cron": "not a cron", + }, + headers=headers, + ) + + assert response.status_code == 204 + assert "cron" in response.headers["HX-Redirect"] + + +def test_the_engine_dashboard_shows_heartbeat_and_status( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """#58's acceptance criterion for the fleet view.""" + login_as(make_user(UserRole.ADMIN)) + session.add(Engine(name="engine-1", token_hash="hashed", version="0.2.0")) + session.commit() + + body = client.get("/engines").text + + assert "engine-1" in body + assert "0.2.0" in body + # No heartbeat yet, and the page says so rather than showing a blank. + assert "never" in body + assert "offline" in body or "active" in body + + +def test_enrolling_an_engine_shows_the_token_exactly_once( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + "/engines/register", + data={"csrf_token": headers["X-CSRF-Token"], "name": "engine-7"}, + headers=headers, + ) + + assert response.status_code == 200 + assert "it is not shown again" in response.text + + engine = session.exec(select(Engine).where(Engine.name == "engine-7")).one() + # Only a hash is stored, and reloading the page cannot show the token again. + assert engine.token_hash + assert engine.token_hash not in response.text + assert "not shown again" not in client.get("/engines").text + + +def test_the_engine_screen_never_renders_token_material( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + login_as(make_user(UserRole.ADMIN)) + engine = Engine(name="engine-2", token_hash="a-stored-hash", version="0.1.0") + session.add(engine) + session.commit() + + body = client.get("/engines").text + + assert "a-stored-hash" not in body + + +# ─── #72 Notification channels & users ─────────────────────────────────────── + + +def test_creating_a_webhook_channel_seals_its_secret( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + headers = login_as(make_user(UserRole.ADMIN)) + + response = client.post( + "/channels", + data={ + "csrf_token": headers["X-CSRF-Token"], + "name": "security-alerts", + "type": "webhook", + "url": "https://chat.example.com/hooks/abc", + "secret": "webhook-signing-secret", + "min_severity": "high", + "enabled": "on", + }, + headers=headers, + ) + + assert response.status_code == 204 + channel = session.exec(select(NotificationChannel)).one() + assert channel.type is NotificationChannelType.WEBHOOK + assert channel.config["url"] == "https://chat.example.com/hooks/abc" + assert channel.event_filter["min_severity"] == "high" + assert channel.config["secret_ref"] + assert "webhook-signing-secret" not in str(channel.config) + + +def test_the_channels_screen_never_renders_the_secret_or_its_ref( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """#72's acceptance criterion: webhook config is never echoed.""" + headers = login_as(make_user(UserRole.ADMIN)) + client.post( + "/channels", + data={ + "csrf_token": headers["X-CSRF-Token"], + "name": "security-alerts", + "type": "webhook", + "url": "https://chat.example.com/hooks/abc", + "secret": "webhook-signing-secret", + }, + headers=headers, + ) + channel = session.exec(select(NotificationChannel)).one() + + body = client.get("/channels").text + + assert "webhook-signing-secret" not in body + assert str(channel.config["secret_ref"]) not in body + assert "sealed" in body + + +@pytest.mark.parametrize("role", [UserRole.VIEWER, UserRole.ANALYST]) +def test_only_an_admin_reaches_the_channels_screen( + client: TestClient, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], + role: UserRole, +) -> None: + headers = login_as(make_user(role)) + + assert client.get("/channels").status_code == 403 + assert ( + client.post( + "/channels", + data={ + "csrf_token": headers["X-CSRF-Token"], + "name": "x", + "type": "webhook", + "url": "https://example.com/hook", + }, + headers=headers, + ).status_code + == 403 + ) + + +def test_deleting_a_channel_is_audited( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + from iceberg_core.models import AuditEvent + + headers = login_as(make_user(UserRole.ADMIN)) + client.post( + "/channels", + data={ + "csrf_token": headers["X-CSRF-Token"], + "name": "security-alerts", + "type": "webhook", + "url": "https://chat.example.com/hooks/abc", + }, + headers=headers, + ) + channel = session.exec(select(NotificationChannel)).one() + + response = client.request("DELETE", f"/channels/{channel.id}", headers=headers) + + assert response.status_code == 204 + actions = { + event.action + for event in session.exec( + select(AuditEvent).where(AuditEvent.target_type == AUDIT_TARGET_CHANNEL) + ) + } + assert actions == {"channel.created", "channel.deleted"} + + +def test_assigning_a_role_from_the_ui_lands_and_is_audited( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """#72's acceptance criterion for user management.""" + from iceberg_core.models import AuditEvent + + headers = login_as(make_user(UserRole.ADMIN)) + target = make_user(UserRole.VIEWER) + + response = client.post( + f"/users/{target.id}", + data={"csrf_token": headers["X-CSRF-Token"], "role": "analyst"}, + headers=headers, + ) + + assert response.status_code == 204 + session.refresh(target) + assert target.role is UserRole.ANALYST + + events = session.exec( + select(AuditEvent) + .where(AuditEvent.target_type == AUDIT_TARGET_USER) + .where(AuditEvent.target_id == target.id) + ).all() + assert [event.action for event in events] == ["user.role_changed"] + assert events[0].from_value == "viewer" + assert events[0].to_value == "analyst" + + +def test_an_admin_cannot_change_their_own_account_from_the_ui( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + admin = make_user(UserRole.ADMIN) + headers = login_as(admin) + + listing = client.get("/users").text + response = client.post( + f"/users/{admin.id}", + data={"csrf_token": headers["X-CSRF-Token"], "role": "viewer"}, + headers=headers, + ) + + assert "that is you" in listing + assert response.status_code == 204 + assert "error=" in response.headers["HX-Redirect"] + + +def test_the_last_admin_cannot_be_demoted_from_the_ui( + client: TestClient, + session: Session, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """Two admins demoting each other must not end with zero.""" + first = make_user(UserRole.ADMIN) + second = make_user(UserRole.ADMIN) + headers = login_as(first) + + client.post( + f"/users/{second.id}", + data={"csrf_token": headers["X-CSRF-Token"], "role": "viewer"}, + headers=headers, + ) + session.refresh(second) + assert second.role is UserRole.VIEWER + + # `first` is now the only admin, and nobody else can demote them. + headers = login_as(second) + response = client.post( + f"/users/{first.id}", + data={"csrf_token": headers["X-CSRF-Token"], "role": "viewer"}, + headers=headers, + ) + assert response.status_code == 403 + + +def test_a_mutation_without_a_csrf_token_is_refused( + client: TestClient, + make_source: Callable[..., Source], + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], +) -> None: + """Cookie auth means the browser attaches credentials cross-site by itself.""" + login_as(make_user(UserRole.ANALYST)) + + response = client.post(f"/sources/{make_source().id}/scan") + + assert response.status_code == 403 + + +def test_an_unknown_record_renders_a_page_not_a_json_body( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + login_as(make_user(UserRole.VIEWER)) + + response = client.get(f"/findings/{uuid.uuid4()}") + + assert response.status_code == 404 + assert response.headers["content-type"].startswith("text/html") + assert "Not found" in response.text diff --git a/apps/api/tests/test_web_shell.py b/apps/api/tests/test_web_shell.py new file mode 100644 index 0000000..6eec66d --- /dev/null +++ b/apps/api/tests/test_web_shell.py @@ -0,0 +1,304 @@ +"""The web shell: authentication gating, role-shaped navigation, headers (#53). + +These are the properties the shell has to hold no matter what any screen does — +that an anonymous visitor never sees data, that the rail offers only what the +signed-in role can actually reach, and that the response carries a policy strict +enough for a page which renders strings out of somebody else's Confluence. +""" + +import re +from collections.abc import Callable + +import pytest +from fastapi.testclient import TestClient +from iceberg_api.web.assets import TEMPLATES_DIR +from iceberg_api.web.security import DOCS_PATHS, build_csp, build_security_headers +from iceberg_core.config import CoreSettings +from iceberg_core.enums import UserRole +from iceberg_core.models import User + +#: Every page a signed-in viewer may open. Admin-only screens are checked +#: separately, because for a viewer they are a 403 rather than a page. +VIEWER_PAGES = ("/", "/findings", "/scans", "/sources", "/schedules", "/suppressions", "/rules") +ADMIN_PAGES = ("/engines", "/channels", "/users") + + +def test_an_anonymous_visitor_is_sent_to_the_login_page(client: TestClient) -> None: + response = client.get("/findings", follow_redirects=False) + + assert response.status_code == 303 + assert response.headers["location"] == "/login?next=/findings" + + +def test_the_login_redirect_remembers_the_query_string(client: TestClient) -> None: + """A bookmarked filter must survive the round trip through the IdP.""" + response = client.get("/findings?state=open&severity=critical", follow_redirects=False) + + assert response.headers["location"] == "/login?next=/findings?state=open&severity=critical" + + +def test_an_htmx_request_without_a_session_gets_a_redirect_header_not_a_body( + client: TestClient, +) -> None: + """A 303 would be followed by fetch and swap a login page into a table cell.""" + response = client.post( + "/findings/00000000-0000-0000-0000-000000000000/triage", + headers={"HX-Request": "true"}, + data={"state": "open"}, + follow_redirects=False, + ) + + assert response.status_code == 204 + assert response.headers["HX-Redirect"].startswith("/login") + assert response.content == b"" + + +@pytest.mark.parametrize("candidate", ["//evil.example", "/\\evil.example", "https://evil.example"]) +def test_a_crafted_next_path_cannot_leave_this_origin(client: TestClient, candidate: str) -> None: + """The login page round-trips `next`; it must never become an open redirect.""" + response = client.get(f"/login?next={candidate}") + + assert response.status_code == 200 + assert "evil.example" not in response.text + + +def test_the_login_page_offers_single_sign_on_and_no_password_field( + client: TestClient, +) -> None: + response = client.get("/login") + + assert response.status_code == 200 + assert "/api/v1/auth/login" in response.text + assert 'type="password"' not in response.text + + +def test_a_signed_in_visitor_is_redirected_away_from_the_login_page( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + login_as(make_user(UserRole.VIEWER)) + + response = client.get("/login", follow_redirects=False) + + assert response.status_code == 303 + assert response.headers["location"] == "/" + + +@pytest.mark.parametrize("path", VIEWER_PAGES) +def test_every_viewer_page_renders_for_a_viewer( + client: TestClient, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], + path: str, +) -> None: + login_as(make_user(UserRole.VIEWER)) + + response = client.get(path) + + assert response.status_code == 200, response.text[:400] + assert response.headers["content-type"].startswith("text/html") + + +@pytest.mark.parametrize("path", ADMIN_PAGES) +def test_the_admin_screens_refuse_a_viewer( + client: TestClient, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], + path: str, +) -> None: + """Hiding the link is a courtesy; require_role is the control.""" + login_as(make_user(UserRole.VIEWER)) + + response = client.get(path) + + assert response.status_code == 403 + assert "Your role does not allow this" in response.text + + +@pytest.mark.parametrize("path", ADMIN_PAGES) +def test_every_admin_screen_renders_for_an_admin( + client: TestClient, + make_user: Callable[..., User], + login_as: Callable[[User], dict[str, str]], + path: str, +) -> None: + login_as(make_user(UserRole.ADMIN)) + + response = client.get(path) + + assert response.status_code == 200, response.text[:400] + + +def test_the_rail_shows_administration_only_to_an_admin( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + """The nav reflects the user's role — the acceptance criterion for #53.""" + login_as(make_user(UserRole.ANALYST)) + analyst_view = client.get("/").text + + login_as(make_user(UserRole.ADMIN)) + admin_view = client.get("/").text + + assert 'href="/engines"' not in analyst_view + assert 'href="/users"' not in analyst_view + assert 'href="/findings"' in analyst_view + + assert 'href="/engines"' in admin_view + assert 'href="/users"' in admin_view + assert 'href="/channels"' in admin_view + + +def test_the_shell_names_the_signed_in_user_and_their_role( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + user = make_user(UserRole.ANALYST) + login_as(user) + + body = client.get("/").text + + assert user.display_name in body + assert "analyst" in body + + +def test_the_shell_carries_the_sessions_csrf_token_for_htmx( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + """Every hx- request inherits it from the shell, so no form can forget one.""" + headers = login_as(make_user(UserRole.ANALYST)) + + body = client.get("/").text + + assert "hx-headers=" in body + assert headers["X-CSRF-Token"] in body + + +def test_logout_clears_the_session_and_returns_to_the_login_page( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + headers = login_as(make_user(UserRole.VIEWER)) + + response = client.post( + "/logout", + data={"csrf_token": headers["X-CSRF-Token"]}, + headers=headers, + follow_redirects=False, + ) + + assert response.status_code == 303 + assert response.headers["location"] == "/login" + assert "iceberg_session" in response.headers.get("set-cookie", "") + + +def test_logout_without_a_csrf_token_is_refused( + client: TestClient, make_user: Callable[..., User], login_as: Callable[[User], dict[str, str]] +) -> None: + login_as(make_user(UserRole.VIEWER)) + + response = client.post("/logout", data={}, follow_redirects=False) + + assert response.status_code == 403 + + +# ─── Security headers ──────────────────────────────────────────────────────── + + +def test_the_policy_forbids_inline_and_eval_script() -> None: + """The whole reason the UI self-hosts Alpine's CSP build.""" + policy = build_csp(is_prod=False) + + assert "script-src 'self'" in policy + assert "'unsafe-eval'" not in policy + assert "script-src 'self' 'unsafe-inline'" not in policy + assert "default-src 'self'" in policy + assert "frame-ancestors 'none'" in policy + assert "object-src 'none'" in policy + + +def test_style_src_keeps_unsafe_inline_only_for_style_attributes() -> None: + """Documented carve-out: inline style= attributes cannot be nonced.""" + policy = build_csp(is_prod=False) + + assert "style-src 'self' 'unsafe-inline'" in policy + + +def test_production_adds_transport_hardening() -> None: + dev = build_security_headers(CoreSettings(environment="dev")) + prod = build_security_headers(CoreSettings(environment="prod")) + + assert "Strict-Transport-Security" not in dev + assert prod["Strict-Transport-Security"].startswith("max-age=31536000") + assert "upgrade-insecure-requests" in prod["Content-Security-Policy"] + assert "upgrade-insecure-requests" not in dev["Content-Security-Policy"] + + +def test_every_response_carries_the_security_headers(client: TestClient) -> None: + response = client.get("/login") + + assert "script-src 'self'" in response.headers["content-security-policy"] + assert response.headers["x-content-type-options"] == "nosniff" + assert response.headers["x-frame-options"] == "DENY" + assert response.headers["referrer-policy"] == "strict-origin-when-cross-origin" + assert response.headers["cross-origin-opener-policy"] == "same-origin" + assert "camera=()" in response.headers["permissions-policy"] + + +def test_the_api_surface_is_covered_by_the_same_policy(client: TestClient) -> None: + """A JSON route can still be navigated to; it gets the headers too.""" + response = client.get("/healthz") + + assert "script-src 'self'" in response.headers["content-security-policy"] + + +@pytest.mark.parametrize("path", sorted(DOCS_PATHS)) +def test_the_interactive_docs_are_exempt(client: TestClient, path: str) -> None: + """Swagger UI loads from a CDN and bootstraps inline; a policy loose enough + to run it would be no policy at all, so it is exempted by path instead.""" + response = client.get(path) + + assert "content-security-policy" not in response.headers + + +# ─── No inline script or style ─────────────────────────────────────────────── + + +#: Jinja comments never reach the browser, and several of them here discuss the +#: very tags this file forbids. Strip them before scanning for markup. +_JINJA_COMMENT = re.compile(r"\{#.*?#\}", re.S) + + +def _markup(path: object) -> str: + return _JINJA_COMMENT.sub("", path.read_text(encoding="utf-8")) # type: ignore[attr-defined] + + +def test_no_template_contains_an_inline_script_or_style_block() -> None: + """The invariant that makes `script-src 'self'` possible at all. + + Checked over the template source rather than a rendered page: a block added + to a screen nobody wrote a test for would otherwise ship, and the CSP would + silently stop it in production instead of here. + """ + offenders: list[str] = [] + for template in sorted(TEMPLATES_DIR.rglob("*.html")): + text = _markup(template) + if "