mirror of
https://github.com/Sea-Haven-Industries/meal-order-manager.git
synced 2026-09-30 22:53:12 +00:00
* feat(api): serve meals on ECS Fargate instead of Lambda Keep the Flask app always-on with in-process jobs so CloudFront no longer fronts a cold-start API Gateway. * fix(jobs): run delayed close and reminder deliveries Wall-clock skip windows dropped the only weekly SQS attempt when Scheduler already fired in Eastern time. Dev schedules stay disabled. * fix(api): return JSON objects and stop logging job payloads Flask now jsonify-s handler dicts so API responses are not HTML, and the worker logs only event and status. * fix(ci): restore the reusable workflow so the required check is named ci / ci Inlining the job reported `ci` instead of the org ruleset's `ci / ci`. * fix(secrets): drop unused os import so ruff check passes * style: apply ruff format so ci-python-app lint passes * fix(infra): give meals its own VPC because prod has none * chore(security): re-key ALB SG checkov suppression after vpc.tf
196 lines
7.5 KiB
Python
196 lines
7.5 KiB
Python
"""Unit tests for the admin API Lambda authorizer (INFRA-100)."""
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
# Make the shared layer importable (conftest also does this, but keep explicit).
|
|
_shared = os.path.join(os.path.dirname(__file__), os.pardir, "src", "shared")
|
|
sys.path.insert(0, os.path.abspath(_shared))
|
|
|
|
# Do NOT set GOOGLE_CLIENT_ID_PARAM at module scope — test_submit_order relies
|
|
# on it defaulting to "" globally. Each test below patches it explicitly.
|
|
os.environ.setdefault("TABLE_NAME", "meal-order-manager-orders-test")
|
|
|
|
_handler_path = os.path.join(
|
|
os.path.dirname(__file__), os.pardir, "src", "server", "admin_authorizer.py"
|
|
)
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"admin_authorizer_handler", os.path.abspath(_handler_path)
|
|
)
|
|
authorizer = importlib.util.module_from_spec(_spec)
|
|
sys.modules["admin_authorizer_handler"] = authorizer
|
|
_spec.loader.exec_module(authorizer)
|
|
|
|
CLIENT_ID = "123456789.apps.googleusercontent.com"
|
|
ADMIN_EMAIL = "adam@seahavenind.com"
|
|
|
|
|
|
def _event(token="good-token"):
|
|
headers = {}
|
|
if token is not None:
|
|
headers["authorization"] = f"Bearer {token}"
|
|
return {"headers": headers}
|
|
|
|
|
|
@patch.dict(
|
|
os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
|
|
)
|
|
@patch("admin_authorizer_handler.get_settings")
|
|
@patch("admin_authorizer_handler._verify_google_token")
|
|
@patch("admin_authorizer_handler._get_google_client_id")
|
|
def test_valid_admin_allowed(mock_cid, mock_verify, mock_settings):
|
|
mock_cid.return_value = CLIENT_ID
|
|
mock_verify.return_value = {"name": "Adam", "email": ADMIN_EMAIL}
|
|
mock_settings.return_value = {"admin_emails": [ADMIN_EMAIL]}
|
|
assert authorizer.lambda_handler(_event(), None) == {"isAuthorized": True}
|
|
|
|
|
|
@patch.dict(
|
|
os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
|
|
)
|
|
@patch("admin_authorizer_handler.get_settings")
|
|
@patch("admin_authorizer_handler._verify_google_token")
|
|
@patch("admin_authorizer_handler._get_google_client_id")
|
|
def test_valid_user_but_not_admin_denied(mock_cid, mock_verify, mock_settings):
|
|
mock_cid.return_value = CLIENT_ID
|
|
mock_verify.return_value = {"name": "Bob", "email": "bob@seahavenind.com"}
|
|
mock_settings.return_value = {"admin_emails": [ADMIN_EMAIL]}
|
|
assert authorizer.lambda_handler(_event(), None) == {"isAuthorized": False}
|
|
|
|
|
|
@patch("admin_authorizer_handler.get_settings")
|
|
@patch(
|
|
"admin_authorizer_handler._verify_portal_token",
|
|
return_value={"name": "Portal Admin", "email": ADMIN_EMAIL},
|
|
)
|
|
@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
|
|
def test_valid_portal_admin_allowed(mock_looks_like, mock_verify, mock_settings):
|
|
mock_settings.return_value = {"admin_emails": [ADMIN_EMAIL]}
|
|
|
|
assert authorizer.lambda_handler(_event("portal-id-token"), None) == {
|
|
"isAuthorized": True
|
|
}
|
|
|
|
|
|
@patch("admin_authorizer_handler.get_settings")
|
|
@patch(
|
|
"admin_authorizer_handler._verify_portal_token",
|
|
return_value={"name": "Portal User", "email": "user@seahavenind.com"},
|
|
)
|
|
@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
|
|
def test_portal_non_admin_denied(mock_looks_like, mock_verify, mock_settings):
|
|
mock_settings.return_value = {"admin_emails": [ADMIN_EMAIL]}
|
|
|
|
assert authorizer.lambda_handler(_event("portal-id-token"), None) == {
|
|
"isAuthorized": False
|
|
}
|
|
|
|
|
|
@patch("admin_authorizer_handler._verify_portal_token", return_value=None)
|
|
@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
|
|
def test_invalid_portal_token_denied_without_google_fallback(
|
|
mock_looks_like, mock_verify
|
|
):
|
|
with patch("admin_authorizer_handler._get_google_client_id") as mock_google:
|
|
assert authorizer.lambda_handler(_event("bad-portal-token"), None) == {
|
|
"isAuthorized": False
|
|
}
|
|
mock_google.assert_not_called()
|
|
|
|
|
|
@patch(
|
|
"admin_authorizer_handler._verify_portal_token",
|
|
side_effect=authorizer.CognitoVerificationUnavailable,
|
|
)
|
|
@patch("admin_authorizer_handler.looks_like_cognito_token", return_value=True)
|
|
def test_portal_verification_outage_denied(mock_looks_like, mock_verify):
|
|
assert authorizer.lambda_handler(_event("portal-id-token"), None) == {
|
|
"isAuthorized": False
|
|
}
|
|
|
|
|
|
@patch.dict(
|
|
os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
|
|
)
|
|
@patch("admin_authorizer_handler._verify_google_token")
|
|
@patch("admin_authorizer_handler._get_google_client_id")
|
|
def test_invalid_token_denied(mock_cid, mock_verify):
|
|
mock_cid.return_value = CLIENT_ID
|
|
mock_verify.return_value = None # bad/expired token or wrong domain
|
|
assert authorizer.lambda_handler(_event("bad"), None) == {"isAuthorized": False}
|
|
|
|
|
|
@patch.dict(
|
|
os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
|
|
)
|
|
def test_missing_token_denied():
|
|
assert authorizer.lambda_handler(_event(token=None), None) == {
|
|
"isAuthorized": False
|
|
}
|
|
|
|
|
|
@patch.dict(
|
|
os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
|
|
)
|
|
@patch("admin_authorizer_handler._get_google_client_id")
|
|
def test_client_id_unavailable_denied(mock_cid):
|
|
mock_cid.return_value = "" # SSM failure or empty -> fail closed
|
|
assert authorizer.lambda_handler(_event(), None) == {"isAuthorized": False}
|
|
|
|
|
|
@patch.dict(os.environ, {"GOOGLE_CLIENT_ID_PARAM": ""}, clear=False)
|
|
def test_authorizer_unconfigured_denied():
|
|
assert authorizer.lambda_handler(_event(), None) == {"isAuthorized": False}
|
|
|
|
|
|
@patch.dict(
|
|
os.environ, {"GOOGLE_CLIENT_ID_PARAM": "/meal-order-manager/google-client-id"}
|
|
)
|
|
@patch("admin_authorizer_handler.get_settings")
|
|
@patch("admin_authorizer_handler._verify_google_token")
|
|
@patch("admin_authorizer_handler._get_google_client_id")
|
|
def test_dynamodb_failure_denied(mock_cid, mock_verify, mock_settings):
|
|
"""A DynamoDB error loading admin_emails fails closed (DENY, not a 500)."""
|
|
mock_cid.return_value = CLIENT_ID
|
|
mock_verify.return_value = {"name": "Adam", "email": ADMIN_EMAIL}
|
|
mock_settings.side_effect = RuntimeError("dynamo down")
|
|
assert authorizer.lambda_handler(_event(), None) == {"isAuthorized": False}
|
|
|
|
|
|
def test_audience_mismatch_denied():
|
|
"""_verify_google_token rejects a token whose aud != configured client ID."""
|
|
with patch.object(authorizer.urllib.request, "urlopen") as mock_open:
|
|
resp = mock_open.return_value.__enter__.return_value
|
|
resp.read.return_value = (
|
|
b'{"aud":"other-client","hd":"seahavenind.com","email":"x@seahavenind.com"}'
|
|
)
|
|
assert authorizer._verify_google_token("t", CLIENT_ID) is None
|
|
|
|
|
|
def test_domain_mismatch_denied():
|
|
"""_verify_google_token rejects a token from a non-allowed Workspace domain."""
|
|
with patch.object(authorizer.urllib.request, "urlopen") as mock_open:
|
|
resp = mock_open.return_value.__enter__.return_value
|
|
resp.read.return_value = (
|
|
f'{{"aud":"{CLIENT_ID}","hd":"evil.com","email":"x@evil.com"}}'.encode()
|
|
)
|
|
assert authorizer._verify_google_token("t", CLIENT_ID) is None
|
|
|
|
|
|
def test_valid_token_decoded():
|
|
with patch.object(authorizer.urllib.request, "urlopen") as mock_open:
|
|
resp = mock_open.return_value.__enter__.return_value
|
|
resp.read.return_value = (
|
|
f'{{"aud":"{CLIENT_ID}","hd":"seahavenind.com",'
|
|
f'"email":"{ADMIN_EMAIL}","name":"Adam"}}'.encode()
|
|
)
|
|
info = authorizer._verify_google_token("t", CLIENT_ID)
|
|
assert info == {"name": "Adam", "email": ADMIN_EMAIL}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(pytest.main([__file__, "-v"]))
|