mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-09-30 11:53:13 +00:00
Some checks are pending
Deploy / deploy (push) Waiting to run
* feat(api): add procurement-api stack - read API + OpenAPI docs page Third CDK stack: API Gateway REST API (IAM SigV4) over both pipelines' tables, replacing SHOC's retired SyncController cross-account DynamoDB scan as the reconciliation/backfill path. - lambdas/api/: handler (healthcheck + docs-token gate + router dispatch), router (single route table), pagination (opaque cursor, hostile -> 400), Decimal-safe serialization, wo_repo/po_repo reads. No VendorReplies. - OpenAPI 3.1 spec as source of truth incl. top-level webhooks section documenting the outbound SHOC feed; phase-2 write endpoints x-planned (router answers 501). Self-contained /docs page, no CDN. - Auth: AWS_IAM on data routes + resource policy scoped to exactly arn:aws:iam::396287094661:role/shoc-backend-dev on GET/*; /docs and /openapi.json carve-out is token-gated in the Lambda via shared web_ui_auth (fail-closed, INFRA-74 posture). - KMS: explicit Decrypt/DescribeKey on the DynamoDB CMK from SSM (name-imported table drops the key association - INFRA-104 class). - Alarms: errors/throttles/duration(p99>=22.5s) + gateway 5xx, ALARM-only to site-alerts. No access logging in v1 (docs ?token= shim stays out of logs); cloud_watch_role=False. - Tests: handler auth-seam + routing + Decimal round-trip; moto cursor pagination incl. hostile cursors; spec<->router drift gate; bundle AST pins for the api command; pytest.ini --cov + loader siblings. - Deploy role: third stack DescribeStacks ARN + procurement-api smoke invoke ARN (re-run create-deploy-role.sh before merge). * harden(api): apply sh-security-review findings to procurement-api Fan-out (6 detectors) + review findings resolved: Correctness / DoS: - pagination: require EXACT key-set match (was subset) so a partial/foreign composite cursor can't reach DynamoDB as an inconsistent ExclusiveStartKey -> ValidationException -> 500; comments Query now pins the cursor's work_order_id to the path entity. - handler: map botocore ValidationException to 400 (defense in depth) so a crafted cursor can't drive the zero-threshold 5xx alarm. - web_ui_auth: compare tokens as bytes; a non-ASCII presented token now fails closed (401) instead of crashing hmac.compare_digest into a 500. Resolves the pre-existing xfail(strict) follow-up test; hardens the web UIs too. Docs page: - typeStr() now escapes the one spec-derived string that reached innerHTML. - spec inlined into the docs <script> block escapes "<" -> < (</script> breakout guard); /openapi.json still served byte-faithful. - Cache-Control: no-store + Referrer-Policy: no-referrer on docs responses so the ?token= URL stays out of caches/Referer. - spec-drift test asserts the committed spec carries no "</" / "<!--". IAM / IaC: - resource policy enumerates the 7 data GET resources instead of GET/* so a future GET route can't silently inherit SHOC cross-account reach. - kms:Decrypt grant gains a kms:ViaService=dynamodb condition. - stage throttling (50 rps / 100 burst) bounds the unauthenticated /docs blast radius below the 10k account default. - corrected the PATCH/POST comment (same-account callers aren't blocked by the resource policy; 501 handler + absent write grant are the gate). - documented the RETAIN log-group first-deploy rollback trap and the resource-policy-needs-redeploy gotcha in-stack. Mandatory GPT-4.1 cross-family review of the full policy surface: no BLOCK/FIX. 675 tests pass, ruff clean, cdk synth green.
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
"""web_ui_auth.is_authenticated coverage (constraint 5c) -- 0% before Phase 8.
|
|
|
|
Auth is the mandatory-review surface, so this pins: fail-closed on an unset ARN
|
|
(read at import -> reload), fail-closed on a Secrets-Manager exception, the TTL
|
|
cache refresh, the Bearer / X-Auth-Token / case-insensitivity header matrix,
|
|
wrong-token, and the non-ASCII-token TypeError (a documenting pin -- the module
|
|
is frozen this phase, so the fix is deferred).
|
|
|
|
Loaded through the shared loader (``load_lambda_module("shared", "web_ui_auth")``)
|
|
and reloaded per test because the secret ARN is read at module-import time.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from tests.support import load_lambda_module
|
|
|
|
_ARN = "arn:aws:secretsmanager:us-east-1:000000000000:secret:web-ui-auth-token"
|
|
|
|
|
|
class _FakeSecrets:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name)
|
|
return {"SecretString": self.value}
|
|
|
|
|
|
class _RaisingSecrets:
|
|
def get_secret_value(self, SecretId): # noqa: N803
|
|
raise RuntimeError("secretsmanager access denied")
|
|
|
|
|
|
@pytest.fixture
|
|
def load_auth(monkeypatch):
|
|
"""Return a loader that (re)imports web_ui_auth with a chosen ARN env, so the
|
|
module-level ``_WEB_UI_AUTH_TOKEN_SECRET_ARN`` (read at import) reflects it."""
|
|
mod = load_lambda_module("shared", "web_ui_auth")
|
|
|
|
def _load(arn=_ARN):
|
|
if arn is None:
|
|
monkeypatch.delenv("WEB_UI_AUTH_TOKEN_SECRET_ARN", raising=False)
|
|
else:
|
|
monkeypatch.setenv("WEB_UI_AUTH_TOKEN_SECRET_ARN", arn)
|
|
# Re-exec the module body in place (importlib.reload can't re-find a
|
|
# file-path-loaded module by name). This re-reads the ARN env and resets
|
|
# _auth_token_cache / _auth_token_cached_at to their module defaults.
|
|
mod.__spec__.loader.exec_module(mod)
|
|
return mod
|
|
|
|
return _load
|
|
|
|
|
|
def _evt(**headers):
|
|
return {"headers": headers}
|
|
|
|
|
|
def test_fail_closed_on_unset_arn(load_auth):
|
|
mod = load_auth(arn=None)
|
|
assert mod.is_authenticated(_evt(**{"x-auth-token": "anything"})) is False
|
|
|
|
|
|
def test_fail_closed_on_secrets_manager_exception(load_auth, monkeypatch):
|
|
mod = load_auth()
|
|
monkeypatch.setattr(mod.boto3, "client", lambda *a, **k: _RaisingSecrets())
|
|
# Fails closed (False) and does NOT raise -- the except in _get_auth_token.
|
|
assert mod.is_authenticated(_evt(**{"x-auth-token": "anything"})) is False
|
|
|
|
|
|
def test_ttl_cache_refresh_picks_up_rotated_secret(load_auth, monkeypatch):
|
|
mod = load_auth()
|
|
fake = _FakeSecrets("tok1")
|
|
monkeypatch.setattr(mod.boto3, "client", lambda *a, **k: fake)
|
|
clock = [1000.0]
|
|
monkeypatch.setattr(mod.time, "monotonic", lambda: clock[0])
|
|
|
|
assert mod._get_auth_token() == "tok1" # first fetch, cached at t=1000
|
|
|
|
fake.value = "tok2" # secret rotated
|
|
clock[0] = 1000.0 + 299 # still within the 300s TTL
|
|
assert mod._get_auth_token() == "tok1" # served from cache
|
|
|
|
clock[0] = 1000.0 + 301 # TTL elapsed
|
|
assert mod._get_auth_token() == "tok2" # refetched, rotation picked up
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("headers", "expected"),
|
|
[
|
|
({"x-auth-token": "SECRET"}, True),
|
|
({"X-Auth-Token": "SECRET"}, True), # header-name case-insensitive
|
|
({"Authorization": "Bearer SECRET"}, True),
|
|
({"authorization": "bearer SECRET"}, True), # scheme case-insensitive
|
|
# X-Auth-Token takes precedence over an Authorization header.
|
|
({"X-Auth-Token": "SECRET", "Authorization": "Bearer WRONG"}, True),
|
|
({"x-auth-token": "WRONG"}, False),
|
|
({"Authorization": "Bearer WRONG"}, False),
|
|
({"Authorization": "SECRET"}, False), # no Bearer prefix
|
|
({}, False), # no credentials at all
|
|
({"x-auth-token": ""}, False), # empty presented token
|
|
],
|
|
)
|
|
def test_header_matrix(load_auth, monkeypatch, headers, expected):
|
|
mod = load_auth()
|
|
monkeypatch.setattr(mod, "_get_auth_token", lambda: "SECRET")
|
|
assert mod.is_authenticated(_evt(**headers)) is expected
|
|
|
|
|
|
def test_non_ascii_token_should_fail_closed(load_auth, monkeypatch):
|
|
# Fixed 2026-07-23 (procurement-api security review): is_authenticated now
|
|
# compares tokens as bytes, so a non-ASCII presented token fails closed to
|
|
# False instead of raising TypeError (which surfaced as a 500 that would
|
|
# page the API's zero-threshold 5xx alarm). Was xfail(strict) while
|
|
# web_ui_auth was frozen; the fix landed with the shared bytes-compare.
|
|
mod = load_auth()
|
|
monkeypatch.setattr(mod, "_get_auth_token", lambda: "SECRET")
|
|
assert mod.is_authenticated(_evt(**{"x-auth-token": "SÉCRET"})) is False
|