mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-09-30 20:03:14 +00:00
121 lines
4.7 KiB
Python
121 lines
4.7 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
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.xfail(
|
||
|
|
strict=True,
|
||
|
|
reason="web_ui_auth is frozen this phase: is_authenticated raises TypeError on "
|
||
|
|
"a non-ASCII presented token (hmac.compare_digest is ASCII-only) and surfaces "
|
||
|
|
"as a 500 instead of failing closed to False. Asserting the DESIRED behavior "
|
||
|
|
"under xfail(strict) documents the intended fix and auto-fails (xpass) the moment "
|
||
|
|
"the module is corrected, forcing the marker's removal. Tracked as a follow-up.",
|
||
|
|
raises=TypeError,
|
||
|
|
)
|
||
|
|
def test_non_ascii_token_should_fail_closed(load_auth, monkeypatch):
|
||
|
|
mod = load_auth()
|
||
|
|
monkeypatch.setattr(mod, "_get_auth_token", lambda: "SECRET")
|
||
|
|
assert mod.is_authenticated(_evt(**{"x-auth-token": "SÉCRET"})) is False
|