From ae04739df7539a8bb2b4a0e3ca66fd636a36460c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 10 Nov 2026 02:34:47 +0000 Subject: [PATCH 1/3] test: exercise the OIDC https key fetch, and add the real-IdP check (#113) #95 tested OIDC with a local key and a file:// key set, so the https:// path (PyJWT's PyJWKClient) ran in no test. These run it for real against a local TLS identity provider: a trusted certificate, key rotation, an untrusted certificate, an IdP that is down, one that hangs, and a full sign-in through the API. - The key fetch now times out after 5 s. PyJWT's default is 30 s, which would hold a request thread that long for every sign-in attempt while the IdP is unreachable. - oidc.verify() keeps the reason a token is rejected; username_from() still hides it. - Corrects a comment: a rotated key is re-fetched at most every 30 s (PyJWT's cooldown), so it can be refused for up to 30 s after rotation. Tested. - scripts/check_oidc.py runs the check against the real IdP, stage by stage, with a hint for each failure and no token or session token ever printed. docs/OIDC_REAL_IDP_CHECK.md is the procedure (device-code flow, Keycloak and Entra). One sign-in against the deployment's real IdP is still needed to close #113; this makes it a single command. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_011T6WqhUfEMZv2Dxerqs2Eq --- api/oidc.py | 26 +++- docs/ENV_VARIABLES.md | 2 +- docs/OIDC_REAL_IDP_CHECK.md | 86 +++++++++++++ docs/TEST_CASES.md | 126 +++++++++++++++++- docs/VCRM.md | 2 +- scripts/check_oidc.py | 199 +++++++++++++++++++++++++++++ tests/oidc_stub.py | 145 +++++++++++++++++++++ tests/system/test_check_oidc.py | 220 ++++++++++++++++++++++++++++++++ tests/system/test_oidc_https.py | 175 +++++++++++++++++++++++++ 9 files changed, 972 insertions(+), 9 deletions(-) create mode 100644 docs/OIDC_REAL_IDP_CHECK.md create mode 100644 scripts/check_oidc.py create mode 100644 tests/oidc_stub.py create mode 100644 tests/system/test_check_oidc.py create mode 100644 tests/system/test_oidc_https.py diff --git a/api/oidc.py b/api/oidc.py index 8de4c39..bfc6c8a 100644 --- a/api/oidc.py +++ b/api/oidc.py @@ -18,6 +18,9 @@ from api.config import settings ALGORITHMS = ["RS256", "ES256"] +# Seconds to wait for the IdP's key set. PyJWT's default is 30, which would hold a request thread that long for +# every sign-in attempt while the IdP is down or unreachable. +JWKS_TIMEOUT_SECONDS = 5 def enabled() -> bool: @@ -36,12 +39,15 @@ def check_config() -> None: @lru_cache(maxsize=4) def _jwks(url: str): - """Key source for ``url``: https:// is fetched (and re-fetched on an unknown - key id, for IdP key rotation); file:// is a JWKS file published inside an + """Key source for ``url``: https:// is fetched; file:// is a JWKS file published inside an enclave. Plain http:// is refused, since keys fetched over it could be swapped. + + An https:// key set is cached for PyJWT's lifespan (5 minutes), and an unknown key id triggers a + re-fetch, for IdP key rotation, but at most once per PyJWT's cooldown (30 seconds): so random key ids + can't be used to hammer the IdP, and a freshly rotated key can be refused for up to 30 seconds. """ if url.startswith("https://"): - return jwt.PyJWKClient(url, cache_keys=True) + return jwt.PyJWKClient(url, cache_keys=True, timeout=JWKS_TIMEOUT_SECONDS) if url.startswith("file://"): return jwt.PyJWKSet.from_json(Path(url2pathname(urlparse(url).path)).read_text(encoding="utf-8")) raise ValueError(f"OIDC_JWKS_URL must be https:// or file://, got {url!r}") @@ -58,12 +64,20 @@ def _signing_key(id_token: str): raise jwt.InvalidTokenError(f"unknown key id {kid!r}") from None +def verify(id_token: str) -> dict: + """The token's claims once its signature, issuer, audience and expiry check out; raises ``jwt.PyJWTError`` if not. + + ``username_from`` hides the reason; this keeps it, for scripts/check_oidc.py. + """ + key = _signing_key(id_token) + return jwt.decode(id_token, key, algorithms=ALGORITHMS, audience=settings.OIDC_AUDIENCE, + issuer=settings.OIDC_ISSUER, options={"require": ["exp", "iat", "iss", "aud"]}) + + def username_from(id_token: str) -> str | None: """The verified token's username claim, or None if anything about it is wrong.""" try: - key = _signing_key(id_token) - claims = jwt.decode(id_token, key, algorithms=ALGORITHMS, audience=settings.OIDC_AUDIENCE, - issuer=settings.OIDC_ISSUER, options={"require": ["exp", "iat", "iss", "aud"]}) + claims = verify(id_token) except jwt.PyJWTError: return None username = claims.get(settings.OIDC_USERNAME_CLAIM) diff --git a/docs/ENV_VARIABLES.md b/docs/ENV_VARIABLES.md index b962785..a0399d2 100644 --- a/docs/ENV_VARIABLES.md +++ b/docs/ENV_VARIABLES.md @@ -47,7 +47,7 @@ Read by `api/config.py`. Set these in the shell before running | `LOGIN_LOCKOUT_MINUTES` | `15` | No | Length of the failure window and of the lock. | | `OIDC_ISSUER` | _(empty)_ | No | Turns on OIDC sign-in (#95). Must match the ID token's `iss`. Leave empty for air-gapped installs that use local accounts. | | `OIDC_AUDIENCE` | _(empty)_ | If OIDC is on | This app's client id; must match the token's `aud`. | -| `OIDC_JWKS_URL` | _(empty)_ | If OIDC is on | The IdP's signing keys: `https://…` (fetched, refreshed when keys rotate) or `file:///…` (a JWKS file published inside an enclave). `http://` is refused. | +| `OIDC_JWKS_URL` | _(empty)_ | If OIDC is on | The IdP's signing keys: `https://…` (fetched, and refreshed when keys rotate: at most every 30 s; a private CA needs `SSL_CERT_FILE`, see `docs/OIDC_REAL_IDP_CHECK.md`) or `file:///…` (a JWKS file published inside an enclave). `http://` is refused. | | `OIDC_USERNAME_CLAIM` | `preferred_username` | No | Token claim that names the local user. The user must already exist and be a member of the tenant; nobody is created from claims. | | `TE_SCHEMA` | `te_dev` | No | Schema holding the 12 fixed T&E tables. Switch to `te_test`, `te_staging`, or `te_prod` per environment. | | `MAX_UPLOAD_BYTES` | `52428800` (50 MB) | No | Maximum file size accepted by the upload endpoint. | diff --git a/docs/OIDC_REAL_IDP_CHECK.md b/docs/OIDC_REAL_IDP_CHECK.md new file mode 100644 index 0000000..a951719 --- /dev/null +++ b/docs/OIDC_REAL_IDP_CHECK.md @@ -0,0 +1,86 @@ +# Checking OIDC sign-in against the real identity provider (#113) + +One sign-in against the deployment's real IdP (Keycloak or Entra ID) proves what no automated test can: that **its** certificate is trusted, **its** issuer and audience match the configuration, and **its** tokens carry a claim that names a local user. This is the procedure and the script for it. It needs access to the IdP, and a person who can sign in there. + +## What is already tested, and what is not + +| | Where | Status | +|---|---|---| +| Signature, issuer, audience, expiry and claim checks | `tests/system/test_oidc.py` (file:// key set) | Automated | +| The `https://` key fetch (PyJWT's `PyJWKClient`), key rotation, an untrusted certificate, an IdP that is down or hangs, the sign-in through the API | `tests/system/test_oidc_https.py`, against a local TLS server | Automated (#113) | +| This script's checks and their hints | `tests/system/test_check_oidc.py` | Automated | +| **That the real IdP's certificate, issuer, audience, claims and key set work with this configuration** | This procedure | **To do, by a person with IdP access** | + +## 1. Gather the values + +| Setting | Keycloak | Entra ID (v2 endpoints) | +|---|---|---| +| `OIDC_ISSUER` | `https:///realms/` | `https://login.microsoftonline.com//v2.0` | +| `OIDC_JWKS_URL` | `/protocol/openid-connect/certs` | `https://login.microsoftonline.com//discovery/v2.0/keys` | +| `OIDC_AUDIENCE` | the client id | the application (client) id | +| `OIDC_USERNAME_CLAIM` | `preferred_username` (default) | `preferred_username` (v2 tokens). v1 tokens use `upn` | + +The IdP publishes the right issuer and key URL at `/.well-known/openid-configuration`; the script compares yours with it. The issuer must match the token's `iss` **exactly**, including any trailing slash. + +## 2. Get an ID token + +Use the device-code flow, so the token never goes through a browser address bar or a third-party site. **Don't paste an ID token into an online debugger, an issue or a chat: it is a credential.** Register a **test** client (public, device flow enabled, scope `openid`), then: + +```bash +umask 077 # the token file is private to you +CLIENT_ID= +# Keycloak: DEVICE=/protocol/openid-connect/auth/device TOKEN=/protocol/openid-connect/token +# Entra: DEVICE=https://login.microsoftonline.com//oauth2/v2.0/devicecode +# TOKEN=https://login.microsoftonline.com//oauth2/v2.0/token +curl -s "$DEVICE" -d "client_id=$CLIENT_ID" -d "scope=openid" | tee /tmp/device.json | jq '{user_code, verification_uri}' +# Sign in at the verification_uri as the user, enter the user_code, then: +curl -s "$TOKEN" -d "client_id=$CLIENT_ID" -d "grant_type=urn:ietf:params:oauth:grant-type:device_code" \ + -d "device_code=$(jq -r .device_code /tmp/device.json)" | jq -r .id_token > idtoken.txt +rm /tmp/device.json +``` + +The user must **already exist** in the platform and be a **member of the tenant** (nobody is created from token claims), and their local username must equal the token's username claim, for example `alice@example.org` for Entra's `preferred_username`. + +## 3. Run the check + +On a machine that can reach the IdP, with the API's `OIDC_*` values and the token file: + +```bash +export OIDC_ISSUER=... OIDC_AUDIENCE=... OIDC_JWKS_URL=... +python scripts/check_oidc.py --token-file idtoken.txt # the IdP, its keys, the token +python scripts/check_oidc.py --token-file idtoken.txt --api http://localhost:8000 --tenant # and the real sign-in +rm idtoken.txt +``` + +A private CA: set `SSL_CERT_FILE=/path/to/ca-bundle.pem` for the script **and** in the API's own environment, or the key fetch fails closed. Passing means every stage printed `OK`. The last line says whether the sign-in itself was checked. The script never prints the token or the session token. + +Run the same sign-in once through the real front end too, because the front end builds the authorization request. + +## 4. What to record on #113 + +Paste the script's output (it holds no secrets), plus: which IdP and version, the date, and who ran it. For example: + +``` +OK config issuer https://…, audience …, keys https://…, username claim 'preferred_username' +OK discovery issuer and key URL match what the IdP publishes +OK keys 2 signing key(s): … +OK token alg RS256, kid …, expires in … s, preferred_username = '…'. Signature, issuer, audience and expiry verified +OK sign-in 200: signed in as '…' to tenant '…' with role '…'; … +``` + +Then close #113. + +## When a stage fails + +| Stage | Message | Usual cause | +|---|---|---| +| keys | `CERTIFICATE_VERIFY_FAILED` | The host doesn't trust the IdP's CA. Set `SSL_CERT_FILE` (above) | +| keys | connection refused or timed out | A firewall or proxy between the API host and the IdP. The API waits 5 s for the key set, then refuses the sign-in | +| discovery | issuer differs | Trailing slash, or the wrong tenant id or realm in `OIDC_ISSUER` | +| token | `InvalidAudienceError` | `OIDC_AUDIENCE` isn't the client id the token was issued for | +| token | `InvalidIssuerError` | The token's `iss` isn't exactly `OIDC_ISSUER` | +| token | no username claim | Set `OIDC_USERNAME_CLAIM` to the claim that holds the username | +| token | key id not in the key set | The IdP rotated its keys. A new key can be refused for up to 30 s after rotation (PyJWT re-fetches at most every 30 s, so random key ids can't be used to hammer the IdP). Retry | +| token | `ImmatureSignatureError` | Clock skew between this host and the IdP | +| sign-in | 401 | The token verifies but that user isn't a member of that tenant, or the user doesn't exist locally | +| sign-in | 404 | The **API's** environment has no `OIDC_ISSUER`: the check passed here, but the API wasn't configured | diff --git a/docs/TEST_CASES.md b/docs/TEST_CASES.md index 675534b..51484c1 100644 --- a/docs/TEST_CASES.md +++ b/docs/TEST_CASES.md @@ -1,7 +1,7 @@ # Test Catalogue > Auto-generated by `scripts/test_catalogue.py` — do not edit by hand. -> 1129 tests across 50 files. +> 1187 tests across 52 files. To regenerate: `python3 scripts/test_catalogue.py` @@ -419,6 +419,78 @@ To regenerate: `python3 scripts/test_catalogue.py` | `test_downgrade_removes_the_triggers_and_columns` | — | — | +## `tests/system/test_check_oidc.py` + +| Test | Markers | Objective | +|---|---|---| +| `EveryStageCanPass.test_a_good_token_passes_every_stage` | — | — | + +| `EveryStageCanPass.test_the_sign_in_stage_reports_the_role_and_never_the_session_token` | — | — | + +| `EachStageNamesItsCause.test_no_oidc_config` | — | — | + +| `EachStageNamesItsCause.test_plain_http_key_url` | — | — | + +| `EachStageNamesItsCause.test_an_issuer_that_differs_from_the_idps_own_is_a_warning_naming_both` | — | — | + +| `EachStageNamesItsCause.test_a_certificate_the_host_does_not_trust_says_what_to_set` | — | — | + +| `EachStageNamesItsCause.test_wrong_audience_expired_and_wrong_issuer_each_say_why` | — | — | + +| `EachStageNamesItsCause.test_a_token_signed_by_someone_else_is_rejected` | — | — | + +| `EachStageNamesItsCause.test_a_missing_username_claim_names_the_setting_to_change` | — | — | + +| `EachStageNamesItsCause.test_something_that_isnt_a_token` | — | — | + +| `EachStageNamesItsCause.test_the_api_refusing_or_not_having_oidc_is_explained` | — | — | + +| `CommandLine.test_success_prints_a_summary_and_exits_zero_without_printing_the_token` | — | — | + +| `CommandLine.test_with_the_api_it_checks_the_sign_in_and_hides_the_session_token` | — | — | + +| `CommandLine.test_failure_exits_one_and_lists_the_failed_stages` | — | — | + +| `CommandLine.test_no_token_is_exit_two_and_the_token_is_not_a_command_line_argument` | — | — | + +| `CommandLine.test_token_from_a_file` | — | — | + +| `CommandLine.test_api_and_tenant_go_together` | — | — | + +| `test_a_good_token_passes_every_stage` | — | — | + +| `test_the_sign_in_stage_reports_the_role_and_never_the_session_token` | — | — | + +| `test_no_oidc_config` | — | — | + +| `test_plain_http_key_url` | — | — | + +| `test_an_issuer_that_differs_from_the_idps_own_is_a_warning_naming_both` | — | — | + +| `test_a_certificate_the_host_does_not_trust_says_what_to_set` | — | — | + +| `test_wrong_audience_expired_and_wrong_issuer_each_say_why` | — | — | + +| `test_a_token_signed_by_someone_else_is_rejected` | — | — | + +| `test_a_missing_username_claim_names_the_setting_to_change` | — | — | + +| `test_something_that_isnt_a_token` | — | — | + +| `test_the_api_refusing_or_not_having_oidc_is_explained` | — | — | + +| `test_success_prints_a_summary_and_exits_zero_without_printing_the_token` | — | — | + +| `test_with_the_api_it_checks_the_sign_in_and_hides_the_session_token` | — | — | + +| `test_failure_exits_one_and_lists_the_failed_stages` | — | — | + +| `test_no_token_is_exit_two_and_the_token_is_not_a_command_line_argument` | — | — | + +| `test_token_from_a_file` | — | — | + +| `test_api_and_tenant_go_together` | — | — | + ## `tests/system/test_failed_uploads.py` | Test | Markers | Objective | @@ -699,6 +771,58 @@ To regenerate: `python3 scripts/test_catalogue.py` | `test_missing_audience_or_insecure_jwks_url_refuses_to_start` | — | — | +## `tests/system/test_oidc_https.py` + +| Test | Markers | Objective | +|---|---|---| +| `KeyFetchOverHttps.test_a_token_is_verified_with_keys_fetched_over_https` | — | — | + +| `KeyFetchOverHttps.test_the_key_set_is_cached_between_sign_ins` | — | — | + +| `KeyFetchOverHttps.test_a_forged_token_is_refused_even_when_it_names_a_published_key` | — | — | + +| `KeyFetchOverHttps.test_claims_are_still_checked_after_the_signature` | — | — | + +| `KeyFetchOverHttps.test_a_rotated_signing_key_is_picked_up_without_a_restart_after_the_cooldown` | — | — | + +| `KeyFetchOverHttps.test_random_key_ids_cannot_make_the_api_hammer_the_idp` | — | — | + +| `WhenTheIdpCannotBeTrustedOrReached.test_a_certificate_the_host_does_not_trust_fails_closed` | — | — | + +| `WhenTheIdpCannotBeTrustedOrReached.test_an_idp_that_is_down_fails_closed_at_once` | — | — | + +| `WhenTheIdpCannotBeTrustedOrReached.test_an_idp_that_hangs_costs_a_sign_in_the_timeout_and_no_more` | — | — | + +| `WhenTheIdpCannotBeTrustedOrReached.test_plain_http_key_urls_are_refused_at_startup` | — | — | + +| `WhenTheIdpCannotBeTrustedOrReached.test_verify_says_why_where_username_from_only_says_no` | — | — | + +| `SignInThroughTheApi.test_the_idp_token_becomes_a_session_with_the_members_role` | — | — | + +| `test_a_token_is_verified_with_keys_fetched_over_https` | — | — | + +| `test_the_key_set_is_cached_between_sign_ins` | — | — | + +| `test_a_forged_token_is_refused_even_when_it_names_a_published_key` | — | — | + +| `test_claims_are_still_checked_after_the_signature` | — | — | + +| `test_a_rotated_signing_key_is_picked_up_without_a_restart_after_the_cooldown` | — | — | + +| `test_random_key_ids_cannot_make_the_api_hammer_the_idp` | — | — | + +| `test_a_certificate_the_host_does_not_trust_fails_closed` | — | — | + +| `test_an_idp_that_is_down_fails_closed_at_once` | — | — | + +| `test_an_idp_that_hangs_costs_a_sign_in_the_timeout_and_no_more` | — | — | + +| `test_plain_http_key_urls_are_refused_at_startup` | — | — | + +| `test_verify_says_why_where_username_from_only_says_no` | — | — | + +| `test_the_idp_token_becomes_a_session_with_the_members_role` | — | — | + ## `tests/system/test_original_bytes.py` | Test | Markers | Objective | diff --git a/docs/VCRM.md b/docs/VCRM.md index c10a802..8b557bc 100644 --- a/docs/VCRM.md +++ b/docs/VCRM.md @@ -155,7 +155,7 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified" | BR-33 | ⚠️ Partial | The staged import (M6, #110, ADR-0002) COPYs the approved rows into a staging table, promotes them, and commits the data, registry row, import record and audit row in one transaction; every imported row carries `_source_row_id` and `_import_id`. Failure injected at three steps leaves nothing behind and the same-key retry then succeeds; a repeat after success returns the original; concurrent imports import once (`test_staged_import.py`, `AllOrNothing`). It works as the restricted runtime role (`test_runtime_role.py`). Open: the direct upload path and T&E mode aren't staged | Integration incl. failure injection | M6 | | BR-34 | ⚠️ Partial | Total, inserted, duplicate and failed counts persisted per import (M1: #90, `tests/test_import_counts.py`); an upload where every row fails is reported as failed and saves nothing, and partial failures are shown on the files list (#123, `tests/test_failed_uploads.py`); over-wide rows reported, not dropped (M1: #88, `tests/test_ragged_rows.py`); excluded and edited rows are now counted: each staged import stores rows in, excluded, edited, duplicate and imported, and the database checks that in = excluded + duplicate + imported (M6, #110, migration 0011, `test_staged_import.py`, `test_migrations.py`). Open: the direct upload path and T&E mode don't record excluded or edited | Integration | M1, M6 | | BR-35 | ✅ | Export formula injection neutralised (M1: #86, `export.test.ts`). Audit log append-only (M3a, #107): triggers reject UPDATE, DELETE and TRUNCATE for every role; actor recorded; hash chain checked by `GET /api/audit/verify` (`test_audit_append_only.py`, 7 tests). The API's runtime role can only read and insert it, and can't disable its triggers (M3d, `test_runtime_role.py`). Reports are filterable on the server (M7, #111): `GET /api/audit/events` filters by actor, action, file, table and time, with a formula-safe CSV download, and `GET /api/csv/uploads` filters and gives each failed attempt's stored reason (`test_reports_and_policy.py`, 23 tests). The frontend audit page doesn't call them yet | DB negative tests + unit | M1, M3, M7 | -| BR-36 | ⚠️ Partial | Done in M2:
• local accounts with lockout (#98, #103)
• OIDC (#105)
• per-tenant schemas, T&E tables included (#100, #104)
• role matrix in `docs/ROLES.md` (#101)

Tests: `tests/test_accounts.py`, `test_tenant_isolation.py`, `test_roles.py`, `test_oidc.py`, `test_password_reset.py`. Each failed with its control removed.
• frontend sign-in; the browser no longer holds the shared key (#115)
• password change, admin reset and break-glass reset (#117; `tests/test_password_reset.py`)

Open: the shared key still grants non-tenant access to scripts and services; security review (#112); real-IdP OIDC check (#113) | Cross-tenant negative tests; role matrix tests | M2 | +| BR-36 | ⚠️ Partial | Done in M2:
• local accounts with lockout (#98, #103)
• OIDC (#105)
• per-tenant schemas, T&E tables included (#100, #104)
• role matrix in `docs/ROLES.md` (#101)

Tests: `tests/test_accounts.py`, `test_tenant_isolation.py`, `test_roles.py`, `test_oidc.py`, `test_password_reset.py`. Each failed with its control removed.
• frontend sign-in; the browser no longer holds the shared key (#115)
• password change, admin reset and break-glass reset (#117; `tests/test_password_reset.py`)

Open: the shared key still grants non-tenant access to scripts and services; security review (#112); real-IdP OIDC check (#113): the https key fetch, key rotation, an untrusted certificate and an IdP that is down or hangs are now tested against a local TLS IdP (`test_oidc_https.py`, 12 tests; the key fetch now times out after 5 s instead of 30), and `scripts/check_oidc.py` with `docs/OIDC_REAL_IDP_CHECK.md` is the check to run once against the real IdP | Cross-tenant negative tests; role matrix tests | M2 | | BR-37 | ⚠️ Partial | Automated now (M7, #111, `docs/SOVEREIGNTY.md`): a static scan of the API and frontend source (`tests/unit/test_sovereignty.py`); the API tests run with non-local connections refused and any attempt failing the run (`make test-egress-blocked`, a CI step; negative control checked); a built-bundle scan, clean on the client and server bundles (`scripts/check_bundle_egress.py`). Classification is marked on every response and the retention is configurable (decision 7). The only external call is the OIDC key fetch, to a deployment-controlled IdP and off unless configured. Open: the offline install dry run on a clean VM (#60); the egress guard covers Python in-process, not libpq or child processes | Egress-blocked run + offline dry run | M7 | | BR-38 | ❌ Not started | No container bundle for the application yet (`infra/Dockerfile` is the schema and evals runner only) | Offline smoke test 7/7 on Windows, macOS and Linux | #128 | diff --git a/scripts/check_oidc.py b/scripts/check_oidc.py new file mode 100644 index 0000000..9b71327 --- /dev/null +++ b/scripts/check_oidc.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Check OIDC sign-in against the deployment's real identity provider (#113, BR-36). + +Run with the same OIDC_* variables the API reads, and a real ID token for a user who is a member of a tenant: + + export OIDC_ISSUER=https://idp.example.org/realms/te OIDC_AUDIENCE=te-migration \ + OIDC_JWKS_URL=https://idp.example.org/realms/te/protocol/openid-connect/certs + export ID_TOKEN=... # or --token-file PATH, or --token-stdin + python scripts/check_oidc.py [--api http://localhost:8000 --tenant acme] + +It runs the checks one at a time, so a failure names its cause: config, discovery, the key set fetched +over https (the path no automated test could run against a real IdP), the token's signature and claims, +and, with --api, the sign-in itself (POST /api/auth/oidc) and the role it returns. The token is never +accepted on the command line (it would sit in shell history) and neither it nor the session token is +printed. Exit status 0 means every check that ran passed. See docs/OIDC_REAL_IDP_CHECK.md. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import jwt # noqa: E402 + +from api import oidc # noqa: E402 +from api.config import settings # noqa: E402 + +TIMEOUT = 10 + + +@dataclass +class Result: + step: str + status: str # "ok", "warn" (worth a look, doesn't fail the run) or "fail" + detail: str + + +def _open(request): + """Open a request with a fresh opener. urlopen() keeps one process-wide opener whose TLS trust is fixed the first + time it is used, so it can't see SSL_CERT_FILE change in the same process.""" + return urllib.request.build_opener().open(request, timeout=TIMEOUT) # nosec B310 — URLs from the operator's own config + + +def _get_json(url: str) -> dict: + with _open(url) as res: + return json.loads(res.read().decode("utf-8")) + + +def check_config() -> Result: + if not settings.OIDC_ISSUER: + return Result("config", "fail", "OIDC_ISSUER is not set") + try: + oidc.check_config() + except RuntimeError as exc: + return Result("config", "fail", str(exc)) + if not settings.OIDC_JWKS_URL.startswith("https://"): + return Result("config", "warn", "OIDC_JWKS_URL is not https://, so the https key fetch isn't being checked") + return Result("config", "ok", f"issuer {settings.OIDC_ISSUER}, audience {settings.OIDC_AUDIENCE}, " + f"keys {settings.OIDC_JWKS_URL}, username claim {settings.OIDC_USERNAME_CLAIM!r}") + + +def check_discovery() -> Result: + url = settings.OIDC_ISSUER.rstrip("/") + "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/.well-known/openid-configuration" + try: + doc = _get_json(url) + except (urllib.error.URLError, OSError, ValueError) as exc: + return Result("discovery", "warn", f"could not read {url}: {exc}. Not required; the other checks still run") + problems = [] + if doc.get("issuer") != settings.OIDC_ISSUER: + problems.append(f"the IdP says its issuer is {doc.get('issuer')!r}, but OIDC_ISSUER is {settings.OIDC_ISSUER!r}" + " (a trailing slash or the wrong tenant id are the usual causes)") + if doc.get("jwks_uri") != settings.OIDC_JWKS_URL: + problems.append(f"the IdP publishes its keys at {doc.get('jwks_uri')!r}, not OIDC_JWKS_URL {settings.OIDC_JWKS_URL!r}") + return Result("discovery", "warn" if problems else "ok", "; ".join(problems) or "issuer and key URL match what the IdP publishes") + + +def check_keys() -> Result: + try: + source = oidc._jwks(settings.OIDC_JWKS_URL) + keys = source.get_signing_keys() if isinstance(source, jwt.PyJWKClient) else list(source.keys) + except Exception as exc: # noqa: BLE001 — any failure here is what the operator needs to see + hint = "" + if "CERTIFICATE_VERIFY_FAILED" in str(exc) or "SSL" in str(exc): + hint = (" The host doesn't trust the IdP's certificate. For a private CA, point SSL_CERT_FILE at the CA bundle " + "(PEM) in the API's environment.") + return Result("keys", "fail", f"{type(exc).__name__}: {exc}.{hint}") + summary = ", ".join(f"{k.key_id or '(no kid)'} [{k.key_type}]" for k in keys) + return Result("keys", "ok" if keys else "fail", f"{len(keys)} signing key(s): {summary}" if keys else "the key set is empty") + + +_HINTS = { + "InvalidAudienceError": "OIDC_AUDIENCE must be this app's client id, and the token's `aud` must contain it", + "InvalidIssuerError": "the token's `iss` must equal OIDC_ISSUER exactly", + "ExpiredSignatureError": "the token has expired; get a fresh one (IdP tokens are short-lived)", + "ImmatureSignatureError": "the token isn't valid yet: check the clocks of this host and the IdP", + "MissingRequiredClaimError": "the token lacks a required claim (exp, iat, iss, aud)", + "InvalidSignatureError": "the signature doesn't match the key the IdP publishes for this key id", + "PyJWKClientError": "the token's key id isn't in the IdP's key set. A freshly rotated key can be refused for up to 30 seconds", + "DecodeError": "this isn't a well-formed ID token (an access token or a truncated paste?)", +} + + +def check_token(token: str) -> Result: + try: + header = jwt.get_unverified_header(token) + claims = jwt.decode(token, options={"verify_signature": False}) + except jwt.PyJWTError as exc: + return Result("token", "fail", f"{type(exc).__name__}: {exc}. {_HINTS['DecodeError']}") + username_claim = settings.OIDC_USERNAME_CLAIM + shown = (f"alg {header.get('alg')}, kid {header.get('kid')}, iss {claims.get('iss')}, aud {claims.get('aud')}, " + f"expires in {int(claims.get('exp', 0) - time.time())} s, {username_claim} = {claims.get(username_claim)!r}") + try: + verified = oidc.verify(token) + except jwt.PyJWTError as exc: + return Result("token", "fail", f"{shown}. Rejected: {type(exc).__name__}: {exc}. {_HINTS.get(type(exc).__name__, '')}".rstrip()) + username = verified.get(username_claim) + if not isinstance(username, str) or not username: + return Result("token", "fail", f"{shown}. Valid, but it has no {username_claim!r} claim, so no user can be named. " + "Set OIDC_USERNAME_CLAIM to the claim that holds the username " + "(Entra ID: preferred_username, or upn in v1 tokens)") + return Result("token", "ok", f"{shown}. Signature, issuer, audience and expiry verified") + + +def check_api(api_url: str, tenant: str, token: str) -> Result: + url = api_url.rstrip("/") + "/api/auth/oidc" + request = urllib.request.Request(url, data=json.dumps({"tenant": tenant, "idToken": token}).encode(), + headers={"Content-Type": "application/json"}, method="POST") + try: + with _open(request) as res: + session = json.loads(res.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + hint = {401: "the token verified here but the API refused it, or that user isn't a member of the tenant", + 404: "the API has OIDC switched off: it needs OIDC_ISSUER set in its own environment"}.get(exc.code, "") + return Result("sign-in", "fail", f"POST {url} gave {exc.code}. {hint}".rstrip()) + except (urllib.error.URLError, OSError) as exc: + return Result("sign-in", "fail", f"could not reach {url}: {exc}") + return Result("sign-in", "ok", f"200: signed in as {session.get('username')!r} to tenant {session.get('tenant')!r} " + f"with role {session.get('role')!r}; session expires {session.get('expiresAt')}") + + +def run_checks(token: str, api_url: str | None = None, tenant: str | None = None, discovery: bool = True) -> list[Result]: + results = [check_config()] + if results[0].status == "fail": + return results + if discovery: + results.append(check_discovery()) + results.append(check_keys()) + results.append(check_token(token)) + if api_url and tenant: + results.append(check_api(api_url, tenant, token)) + return results + + +def read_token(args) -> str: + if args.token_stdin: + return sys.stdin.read().strip() + if args.token_file: + return Path(args.token_file).read_text(encoding="utf-8").strip() + return os.environ.get(args.token_env, "").strip() + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("--token-env", default="ID_TOKEN", help="environment variable holding the ID token (default ID_TOKEN)") + ap.add_argument("--token-file", help="a file holding the ID token") + ap.add_argument("--token-stdin", action="store_true", help="read the ID token from standard input") + ap.add_argument("--api", help="also sign in through a running API, e.g. http://localhost:8000") + ap.add_argument("--tenant", help="the tenant slug to sign in to (with --api)") + ap.add_argument("--no-discovery", action="store_true", help="skip reading the IdP's discovery document") + args = ap.parse_args(argv) + if bool(args.api) != bool(args.tenant): + ap.error("--api and --tenant go together") + token = read_token(args) + if not token: + print(f"No ID token: set {args.token_env}, or use --token-file or --token-stdin.", file=sys.stderr) + return 2 + results = run_checks(token, args.api, args.tenant, discovery=not args.no_discovery) + width = max(len(r.step) for r in results) + for r in results: + print(f"{r.status.upper():<5} {r.step:<{width}} {r.detail}") + failed = [r.step for r in results if r.status == "fail"] + ran_api = any(r.step == "sign-in" for r in results) + print("\nFAILED: " + ", ".join(failed) if failed else + "\nOK: " + ("sign-in works end to end." if ran_api else + "the token verifies. Add --api and --tenant to check the sign-in itself.")) + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/oidc_stub.py b/tests/oidc_stub.py new file mode 100644 index 0000000..13fccd2 --- /dev/null +++ b/tests/oidc_stub.py @@ -0,0 +1,145 @@ +"""A local identity provider for tests: a real TLS server publishing a JWKS (#113, BR-36). + +``LocalIdp`` serves ``https://localhost:/jwks`` with a throwaway self-signed certificate, so the +API's ``https://`` key fetch (PyJWT's PyJWKClient) runs for real without leaving the machine. Trust the +certificate with ``trust()``, which points ``SSL_CERT_FILE`` at it. Keys can be swapped while it runs, to +rehearse an IdP rotating its signing keys. ``blackhole()`` is a port that accepts a connection and never +answers, for an IdP that hangs. +""" + +from __future__ import annotations + +import contextlib +import datetime +import ipaddress +import json +import os +import socket +import ssl +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest.mock import patch + +import jwt +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + + +def new_key(): + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +def jwk_for(private_key, kid: str) -> dict: + jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(private_key.public_key())) + jwk.update({"kid": kid, "use": "sig", "alg": "RS256"}) + return jwk + + +def _self_signed(directory: Path) -> tuple[Path, Path]: + key = new_key() + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)).not_valid_after(now + datetime.timedelta(days=1)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName("localhost"), + x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), critical=False) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .sign(key, hashes.SHA256())) + cert_path, key_path = directory / "idp.pem", directory / "idp.key" + cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + key_path.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption())) + return cert_path, key_path + + +class LocalIdp: + def __init__(self, keys: dict[str, object] | None = None): + self.keys = dict(keys or {}) # kid -> private key; what /jwks publishes + self.requests: list[str] = [] + self._dir = tempfile.TemporaryDirectory() + self.cert_path, key_path = _self_signed(Path(self._dir.name)) + outer = self + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): # noqa: N802 — http.server's name + outer.requests.append(self.path) + if self.path == "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/.well-known/openid-configuration": + body = json.dumps({"issuer": outer.issuer, "jwks_uri": outer.jwks_url}).encode() + elif self.path == "/jwks": + body = json.dumps({"keys": [jwk_for(k, kid) for kid, k in outer.keys.items()]}).encode() + else: + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(self.cert_path, key_path) + self._server.socket = context.wrap_socket(self._server.socket, server_side=True) + self.port = self._server.server_address[1] + self.jwks_url = f"https://localhost:{self.port}/jwks" + self.issuer = f"https://localhost:{self.port}" # discovery is at /.well-known/openid-configuration + threading.Thread(target=self._server.serve_forever, daemon=True).start() + + def trust(self): + """Make the default TLS context (what PyJWKClient uses) accept this server's certificate.""" + return patch.dict(os.environ, {"SSL_CERT_FILE": str(self.cert_path)}) + + def close(self): + self._server.shutdown() + self._server.server_close() + self._dir.cleanup() + + def __enter__(self): + return self + + def __exit__(self, *exc): + self.close() + + +@contextlib.contextmanager +def blackhole(): + """A loopback port that accepts connections and never answers. Yields its https URL.""" + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen(8) + held: list[socket.socket] = [] + stop = threading.Event() + + def accept(): + listener.settimeout(0.2) + while not stop.is_set(): + with contextlib.suppress(OSError): + held.append(listener.accept()[0]) + + threading.Thread(target=accept, daemon=True).start() + try: + yield f"https://localhost:{listener.getsockname()[1]}/jwks" + finally: + stop.set() + time.sleep(0.25) + for conn in held: + conn.close() + listener.close() + + +def closed_port_url() -> str: + """An https URL on a loopback port nothing is listening on.""" + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() + return f"https://localhost:{port}/jwks" diff --git a/tests/system/test_check_oidc.py b/tests/system/test_check_oidc.py new file mode 100644 index 0000000..95ed1c7 --- /dev/null +++ b/tests/system/test_check_oidc.py @@ -0,0 +1,220 @@ +"""scripts/check_oidc.py, the real-IdP check (#113), run against a local TLS identity provider. + +Each stage can fail on its own and says why, and neither the ID token nor the session token is ever printed. +""" + +from __future__ import annotations + +import contextlib +import importlib.util +import io +import json +import os +import sys +import threading +import time +import unittest +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from unittest.mock import patch + +import jwt + +from api import oidc +from api.config import settings +from tests.oidc_stub import LocalIdp, new_key + +_AUD = "te-migration" +_SCRIPT = Path(__file__).resolve().parent.parent.parent / "scripts" / "check_oidc.py" + + +def _script(): + spec = importlib.util.spec_from_file_location("check_oidc", _SCRIPT) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module # @dataclass looks its module up here + spec.loader.exec_module(module) + return module + + +def _token(key, kid, **overrides) -> str: + now = int(time.time()) + claims = {"aud": _AUD, "iat": now, "exp": now + 300, "sub": "abc", "preferred_username": "alice"} + claims.update(overrides) + return jwt.encode({k: v for k, v in claims.items() if v is not None}, key, algorithm="RS256", headers={"kid": kid}) + + +class _ApiStub: + """A plain-http stand-in for the API's POST /api/auth/oidc.""" + + def __init__(self, status: int = 200): + outer = self + self.seen: list[dict] = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): # noqa: N802 + outer.seen.append(json.loads(self.rfile.read(int(self.headers["Content-Length"])))) + body = json.dumps({"token": "SESSION-SECRET-TOKEN", "username": "alice", "tenant": "acme", "role": "reviewer", + "expiresAt": "2026-11-10T00:00:00+00:00"} if status == 200 else {"detail": "no"}).encode() + self.send_response(status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + self.server = HTTPServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self.server.server_address[1]}" + threading.Thread(target=self.server.serve_forever, daemon=True).start() + + def close(self): + self.server.shutdown() + self.server.server_close() + + +class _Base(unittest.TestCase): + def setUp(self): + self.script = _script() + self._orig = (settings.OIDC_ISSUER, settings.OIDC_AUDIENCE, settings.OIDC_JWKS_URL, settings.OIDC_USERNAME_CLAIM) + oidc._jwks.cache_clear() + self.key = new_key() + self.idp = LocalIdp({"k1": self.key}) + self.addCleanup(self.idp.close) + settings.OIDC_ISSUER, settings.OIDC_AUDIENCE, settings.OIDC_JWKS_URL = self.idp.issuer, _AUD, self.idp.jwks_url + settings.OIDC_USERNAME_CLAIM = "preferred_username" + self.trust = self.idp.trust() + self.trust.__enter__() + + def tearDown(self): + self.trust.__exit__(None, None, None) + settings.OIDC_ISSUER, settings.OIDC_AUDIENCE, settings.OIDC_JWKS_URL, settings.OIDC_USERNAME_CLAIM = self._orig + oidc._jwks.cache_clear() + + def token(self, **overrides) -> str: + overrides.setdefault("iss", self.idp.issuer) + return _token(self.key, "k1", **overrides) + + def by_step(self, results): + return {r.step: r for r in results} + + +class EveryStageCanPass(_Base): + def test_a_good_token_passes_every_stage(self): + results = self.script.run_checks(self.token()) + self.assertEqual([(r.step, r.status) for r in results], + [("config", "ok"), ("discovery", "ok"), ("keys", "ok"), ("token", "ok")], results) + self.assertIn("1 signing key(s): k1", results[2].detail) + self.assertIn("preferred_username = 'alice'", results[3].detail) + + def test_the_sign_in_stage_reports_the_role_and_never_the_session_token(self): + api = _ApiStub() + self.addCleanup(api.close) + results = self.script.run_checks(self.token(), api.url, "acme") + self.assertEqual(results[-1].status, "ok", results[-1]) + self.assertIn("role 'reviewer'", results[-1].detail) + self.assertNotIn("SESSION-SECRET-TOKEN", results[-1].detail) + self.assertEqual(api.seen[0]["tenant"], "acme") + + +class EachStageNamesItsCause(_Base): + def test_no_oidc_config(self): + settings.OIDC_ISSUER = "" + self.assertEqual([(r.step, r.status) for r in self.script.run_checks("x")], [("config", "fail")]) + + def test_plain_http_key_url(self): + settings.OIDC_JWKS_URL = "http://idp.example.test/jwks" + self.assertIn("https:// or file://", self.script.run_checks("x")[0].detail) + + def test_an_issuer_that_differs_from_the_idps_own_is_a_warning_naming_both(self): + settings.OIDC_ISSUER = self.idp.issuer + "/" # the classic trailing slash; discovery lives at the same base + discovery = self.by_step(self.script.run_checks(self.token()))["discovery"] + self.assertEqual(discovery.status, "warn") + self.assertIn("trailing slash", discovery.detail) + + def test_a_certificate_the_host_does_not_trust_says_what_to_set(self): + with patch.dict(os.environ): # restored on exit + os.environ.pop("SSL_CERT_FILE", None) # the host doesn't know this IdP's CA + keys = self.by_step(self.script.run_checks(self.token(), discovery=False))["keys"] + self.assertEqual(keys.status, "fail") + self.assertIn("SSL_CERT_FILE", keys.detail) + + def test_wrong_audience_expired_and_wrong_issuer_each_say_why(self): + now = int(time.time()) + cases = {"OIDC_AUDIENCE": self.token(aud="another-app"), "expired": self.token(exp=now - 5, iat=now - 400), + "`iss` must equal": self.token(iss="https://evil.example/realms/te")} + for expected, token in cases.items(): + result = self.by_step(self.script.run_checks(token, discovery=False))["token"] + self.assertEqual(result.status, "fail", expected) + self.assertIn(expected, result.detail) + + def test_a_token_signed_by_someone_else_is_rejected(self): + forged = _token(new_key(), "k1", iss=self.idp.issuer) + result = self.by_step(self.script.run_checks(forged, discovery=False))["token"] + self.assertEqual(result.status, "fail") + self.assertIn("InvalidSignatureError", result.detail) + + def test_a_missing_username_claim_names_the_setting_to_change(self): + result = self.by_step(self.script.run_checks(self.token(preferred_username=None), discovery=False))["token"] + self.assertEqual(result.status, "fail") + self.assertIn("OIDC_USERNAME_CLAIM", result.detail) + + def test_something_that_isnt_a_token(self): + result = self.by_step(self.script.run_checks("not-a-token", discovery=False))["token"] + self.assertEqual(result.status, "fail") + self.assertIn("well-formed", result.detail) + + def test_the_api_refusing_or_not_having_oidc_is_explained(self): + for status, expected in ((401, "isn't a member"), (404, "OIDC_ISSUER set in its own environment")): + api = _ApiStub(status) + self.addCleanup(api.close) + result = self.script.check_api(api.url, "acme", self.token()) + self.assertEqual(result.status, "fail") + self.assertIn(str(status), result.detail) + self.assertIn(expected, result.detail) + + +class CommandLine(_Base): + def run_main(self, *argv, env=None): + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err), patch.dict(os.environ, env or {}): + code = self.script.main(list(argv)) + return code, out.getvalue(), err.getvalue() + + def test_success_prints_a_summary_and_exits_zero_without_printing_the_token(self): + token = self.token() + code, out, _ = self.run_main(env={"ID_TOKEN": token}) + self.assertEqual(code, 0, out) + self.assertIn("OK: the token verifies", out) + self.assertNotIn(token, out) + self.assertNotIn(token.split(".")[1], out, "not even its payload") + + def test_with_the_api_it_checks_the_sign_in_and_hides_the_session_token(self): + api = _ApiStub() + self.addCleanup(api.close) + code, out, _ = self.run_main("--api", api.url, "--tenant", "acme", env={"ID_TOKEN": self.token()}) + self.assertEqual(code, 0, out) + self.assertIn("sign-in works end to end", out) + self.assertNotIn("SESSION-SECRET-TOKEN", out) + + def test_failure_exits_one_and_lists_the_failed_stages(self): + code, out, _ = self.run_main(env={"ID_TOKEN": self.token(aud="another-app")}) + self.assertEqual(code, 1) + self.assertIn("FAILED: token", out) + + def test_no_token_is_exit_two_and_the_token_is_not_a_command_line_argument(self): + code, _, err = self.run_main(env={"ID_TOKEN": ""}) + self.assertEqual(code, 2) + self.assertIn("ID_TOKEN", err) + with self.assertRaises(SystemExit): + self.run_main("--token", "abc") # no such option: a token on the command line is refused + + def test_token_from_a_file(self): + path = Path(os.environ.get("TMPDIR", "/tmp")) / f"idtoken_{os.getpid()}.txt" + path.write_text(self.token() + "\n") + self.addCleanup(path.unlink) + code, out, _ = self.run_main("--token-file", str(path), env={"ID_TOKEN": ""}) + self.assertEqual(code, 0, out) + + def test_api_and_tenant_go_together(self): + with self.assertRaises(SystemExit): + self.run_main("--api", "http://localhost:8000", env={"ID_TOKEN": "x"}) diff --git a/tests/system/test_oidc_https.py b/tests/system/test_oidc_https.py new file mode 100644 index 0000000..a464438 --- /dev/null +++ b/tests/system/test_oidc_https.py @@ -0,0 +1,175 @@ +"""OIDC key fetch over https:// (#113, BR-36). + +#95's tests used a local RSA key and a file:// JWKS, so the https:// path (PyJWT's PyJWKClient) ran in no +test. These run it for real against a local TLS server (tests/oidc_stub.py): a trusted certificate, +key rotation, an untrusted certificate, an IdP that is down and one that hangs. They don't replace one +sign-in against the deployment's real IdP (docs/OIDC_REAL_IDP_CHECK.md), which proves its certificate, +issuer, audience and claims. +""" + +from __future__ import annotations + +import os +import time +import unittest +import uuid +from unittest.mock import patch + +import jwt +from fastapi.testclient import TestClient + +from api import oidc +from api.config import settings +from api.db import Conn +from api.main import app +from api.services import accounts +from tests.oidc_stub import LocalIdp, blackhole, closed_port_url, new_key + +_ISS, _AUD = "https://idp.example.test/realms/te", "te-migration" + + +def _token(key, kid, username="alice", **overrides) -> str: + now = int(time.time()) + claims = {"iss": _ISS, "aud": _AUD, "iat": now, "exp": now + 300, "sub": "abc", "preferred_username": username} + claims.update(overrides) + return jwt.encode({k: v for k, v in claims.items() if v is not None}, key, algorithm="RS256", headers={"kid": kid}) + + +class _Https(unittest.TestCase): + def setUp(self): + self._orig = (settings.OIDC_ISSUER, settings.OIDC_AUDIENCE, settings.OIDC_JWKS_URL) + settings.OIDC_ISSUER, settings.OIDC_AUDIENCE = _ISS, _AUD + oidc._jwks.cache_clear() + + def tearDown(self): + settings.OIDC_ISSUER, settings.OIDC_AUDIENCE, settings.OIDC_JWKS_URL = self._orig + oidc._jwks.cache_clear() + + +class KeyFetchOverHttps(_Https): + def test_a_token_is_verified_with_keys_fetched_over_https(self): + key = new_key() + with LocalIdp({"k1": key}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + oidc.check_config() + self.assertEqual(oidc.username_from(_token(key, "k1")), "alice") + self.assertEqual(idp.requests, ["/jwks"], "the key set was fetched once, over TLS") + + def test_the_key_set_is_cached_between_sign_ins(self): + key = new_key() + with LocalIdp({"k1": key}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + for _ in range(3): + self.assertEqual(oidc.username_from(_token(key, "k1")), "alice") + self.assertEqual(len(idp.requests), 1) + + def test_a_forged_token_is_refused_even_when_it_names_a_published_key(self): + with LocalIdp({"k1": new_key()}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + self.assertIsNone(oidc.username_from(_token(new_key(), "k1"))) + + def test_claims_are_still_checked_after_the_signature(self): + key = new_key() + with LocalIdp({"k1": key}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + for name, token in {"audience": _token(key, "k1", aud="another-app"), + "issuer": _token(key, "k1", iss="https://evil.example/realms/te"), + "expiry": _token(key, "k1", exp=int(time.time()) - 5, iat=int(time.time()) - 400)}.items(): + self.assertIsNone(oidc.username_from(token), name) + + def test_a_rotated_signing_key_is_picked_up_without_a_restart_after_the_cooldown(self): + old, new = new_key(), new_key() + with LocalIdp({"old": old}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + self.assertEqual(oidc.username_from(_token(old, "old")), "alice") + client = oidc._jwks(idp.jwks_url) + client.cooldown_duration = 0.5 # PyJWT's default is 30 s; shortened to keep the test quick + idp.keys = {"old": old, "new": new} # the IdP rotates + self.assertIsNone(oidc.username_from(_token(new, "new")), "refused inside the cooldown after the last fetch") + self.assertEqual(len(idp.requests), 1, "...and that didn't cost the IdP a request") + time.sleep(0.6) + self.assertEqual(oidc.username_from(_token(new, "new")), "alice", "after the cooldown the key set is re-fetched") + self.assertEqual(oidc.username_from(_token(old, "old")), "alice", "the old key still works during the overlap") + self.assertEqual(len(idp.requests), 2) + + def test_random_key_ids_cannot_make_the_api_hammer_the_idp(self): + key = new_key() + with LocalIdp({"k1": key}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + self.assertEqual(oidc.username_from(_token(key, "k1")), "alice") + for i in range(25): + self.assertIsNone(oidc.username_from(_token(new_key() if i == 0 else key, f"made-up-{i}"))) + self.assertEqual(len(idp.requests), 1, "the unknown ids were refused without re-fetching") + + +class WhenTheIdpCannotBeTrustedOrReached(_Https): + def test_a_certificate_the_host_does_not_trust_fails_closed(self): + key = new_key() + with LocalIdp({"k1": key}) as idp: # no idp.trust(): self-signed and unknown + settings.OIDC_JWKS_URL = idp.jwks_url + self.assertIsNone(oidc.username_from(_token(key, "k1"))) + self.assertEqual(idp.requests, [], "the TLS handshake failed, so no request was made") + + def test_an_idp_that_is_down_fails_closed_at_once(self): + settings.OIDC_JWKS_URL = closed_port_url() + started = time.monotonic() + self.assertIsNone(oidc.username_from(_token(new_key(), "k1"))) + self.assertLess(time.monotonic() - started, 3) + + def test_an_idp_that_hangs_costs_a_sign_in_the_timeout_and_no_more(self): + self.assertEqual(oidc.JWKS_TIMEOUT_SECONDS, 5, "PyJWT's default would be 30") + with blackhole() as url, patch.object(oidc, "JWKS_TIMEOUT_SECONDS", 1): + settings.OIDC_JWKS_URL = url + started = time.monotonic() + self.assertIsNone(oidc.username_from(_token(new_key(), "k1"))) + self.assertLess(time.monotonic() - started, 4) + + def test_plain_http_key_urls_are_refused_at_startup(self): + settings.OIDC_JWKS_URL = "http://idp.example.test/jwks" + with self.assertRaisesRegex(RuntimeError, "https:// or file://"): + oidc.check_config() + + def test_verify_says_why_where_username_from_only_says_no(self): + key = new_key() + with LocalIdp({"k1": key}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + with self.assertRaises(jwt.InvalidAudienceError): + oidc.verify(_token(key, "k1", aud="another-app")) + self.assertEqual(oidc.verify(_token(key, "k1"))["preferred_username"], "alice") + + +class SignInThroughTheApi(_Https): + @classmethod + def setUpClass(cls): + cls.ctx = TestClient(app) + cls.client = cls.ctx.__enter__() + + @classmethod + def tearDownClass(cls): + cls.ctx.__exit__(None, None, None) + + def test_the_idp_token_becomes_a_session_with_the_members_role(self): + tag = f"{os.getpid()}_{uuid.uuid4().hex[:8]}" + tenant, user = f"oh_{tag}", f"oidc_https_{tag}" + with Conn() as conn, conn.cursor() as cur: + tid = accounts.create_tenant(cur, tenant, "OIDC https") + accounts.add_membership(cur, accounts.create_user(cur, user, "unused unused unused"), tid, "reviewer") + conn.commit() + key = new_key() + with LocalIdp({"k1": key}) as idp, idp.trust(): + settings.OIDC_JWKS_URL = idp.jwks_url + res = self.client.post("/api/auth/oidc", json={"tenant": tenant, "idToken": _token(key, "k1", username=user)}) + self.assertEqual(res.status_code, 200, res.text) + self.assertEqual((res.json()["username"], res.json()["role"], res.json()["tenant"]), (user, "reviewer", tenant)) + stranger = self.client.post("/api/auth/oidc", json={"tenant": tenant, "idToken": _token(key, "k1", username="nobody")}) + self.assertEqual(stranger.status_code, 401, "a valid token for someone who isn't a member gets nothing") + forged = self.client.post("/api/auth/oidc", json={"tenant": tenant, "idToken": _token(new_key(), "k1", username=user)}) + self.assertEqual(forged.status_code, 401) + with Conn() as conn, conn.cursor() as cur: + from psycopg2 import sql + from api.tenancy import schema_for + control = sql.Identifier(settings.CONTROL_SCHEMA) + cur.execute(sql.SQL("DELETE FROM {}.tenants WHERE slug = %s").format(control), (tenant,)) + cur.execute(sql.SQL("DELETE FROM {}.users WHERE username = %s").format(control), (user,)) + cur.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema_for(tenant)))) + conn.commit() From ff3f789798575cd418901ac142ff8cbe8a2338ba Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 10 Nov 2026 02:46:08 +0000 Subject: [PATCH 2/3] docs: security review pack for #112, with the author's own pass and its evidence Preparation for the independent review ADR-0003 asks for before the first external tenant; not the review. docs/SECURITY_REVIEW_PACK.md maps the five scoped areas (authentication, tenant isolation, role matrix, upload handling, deployment secrets) to their controls, code and tests, lists the questions for the reviewer, and records 11 weaknesses found while preparing it (F1 to F11), each with how it was verified. - make security-evidence runs the tests behind it (231 pass). - The print window in the frontend is the one hand-built HTML sink and had no test; export.test.ts now checks that every user-supplied value is escaped there. - A CI job runs the frontend vitest suite, which no job ran before. - pip-audit and npm audit (production) were clean on 2026-11-10. Nothing found has been fixed: whether to fix, accept or defer each is the review's output. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_011T6WqhUfEMZv2Dxerqs2Eq --- .github/workflows/quality-gate.yml | 24 +++ Makefile | 12 +- .../src/__tests__/export.test.ts | 64 ++++++++ docs/SECURITY_REVIEW_PACK.md | 145 ++++++++++++++++++ docs/VCRM.md | 2 +- 5 files changed, 245 insertions(+), 2 deletions(-) create mode 100644 docs/SECURITY_REVIEW_PACK.md diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index f9f2f71..29ec3b2 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -200,6 +200,30 @@ jobs: if-no-files-found: ignore retention-days: 30 + # #112: the frontend's own unit tests (vitest). No other job ran them, so a change to the one place the app + # builds HTML by hand (the print window in src/lib/export.ts) could lose its escaping without anything failing. + frontend-tests: + name: frontend tests (vitest) + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v7 + + - name: Setup Node + uses: actions/setup-node@v6 + with: + node-version: '22' + cache: 'npm' + cache-dependency-path: csv-table-hub-main/package-lock.json + + - name: Install frontend dependencies + working-directory: csv-table-hub-main + run: npm ci + + - name: Frontend unit tests + working-directory: csv-table-hub-main + run: npm test + # #168: WCAG 2.1 A and AA scans of the frontend (csv-table-hub-main) with # Playwright + axe-core in Chromium, against the Vite dev server. The API is # stubbed, so this job needs no database and a backend-only change cannot diff --git a/Makefile b/Makefile index b0faa6a..5eb3bca 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ .PHONY: test-free test-gate test-evals test-e2e test-all \ - test-unit test-system test-egress-blocked check-bundle test-sit test-performance test-accessibility test-uat \ + test-unit test-system test-egress-blocked check-bundle security-evidence test-sit test-performance test-accessibility test-uat \ lint lint-diff health \ eval-list eval-compare eval-summary select-tests \ csv-load csv-list csv-demo @@ -60,6 +60,16 @@ test-egress-blocked: check-bundle: python3 scripts/check_bundle_egress.py csv-table-hub-main/.output +# The tests behind docs/SECURITY_REVIEW_PACK.md (#112): authentication, tenant isolation, roles, upload handling and the +# audit trail. Needs PostgreSQL. A passing run is evidence for the reviewer, not the review. +security-evidence: + python3 -m pytest tests/system/test_accounts.py tests/system/test_password_reset.py tests/system/test_oidc.py \ + tests/system/test_oidc_https.py tests/system/test_check_oidc.py tests/system/test_tenant_isolation.py \ + tests/system/test_tenant_te_migrations.py tests/system/test_roles.py tests/system/test_runtime_role.py \ + tests/system/test_audit_append_only.py tests/system/test_preserved_uploads.py tests/system/test_staged_import.py \ + tests/system/test_versioned_datasets.py tests/system/test_reports_and_policy.py \ + tests/unit/test_security.py tests/unit/test_api_hardening.py tests/unit/test_sovereignty.py -v + test-sit: $(call run_category,tests/system_integration) diff --git a/csv-table-hub-main/src/__tests__/export.test.ts b/csv-table-hub-main/src/__tests__/export.test.ts index 2f93ad5..73be0ea 100644 --- a/csv-table-hub-main/src/__tests__/export.test.ts +++ b/csv-table-hub-main/src/__tests__/export.test.ts @@ -305,3 +305,67 @@ describe("downloadAuditReport", () => { expect(text).toContain("Imported"); }); }); + +// ─── print window (PDF): the one place HTML is built by hand ────────────────── +// +// The print window is written with document.write(), not React, so nothing escapes for us, and +// it shares the app's origin (and so its sessionStorage, which holds the sign-in token). File +// names, cell values and reasons all come from uploaded CSVs. Security review pack, #112. + +describe("print window HTML (PDF) escapes everything user-supplied", () => { + const HOSTILE = [ + ``, + `">`, + `'-alert(1)-'`, + `<already> & `, + ]; + + let written: string; + + beforeEach(() => { + written = ""; + vi.useFakeTimers(); + vi.spyOn(window, "open").mockImplementation(() => ({ + document: { write: (html: string) => { written += html; }, close: vi.fn() }, + focus: vi.fn(), + print: vi.fn(), + close: vi.fn(), + }) as unknown as Window); + }); + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + function assertNothingExecutable() { + expect(written).not.toMatch(/