File indexing completed on 2026-11-09 10:47:22
0001
0002 """Check the SWF proxy contract without network or database writes."""
0003
0004 import asyncio
0005 from copy import deepcopy
0006 from datetime import datetime, timedelta, timezone
0007 import hashlib
0008 import logging
0009 import os
0010 from pathlib import Path
0011 import sys
0012 import tempfile
0013 from unittest.mock import AsyncMock, patch
0014
0015 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
0016 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "swf_monitor_project.settings")
0017
0018 import django
0019 django.setup()
0020 logging.getLogger("httpx").setLevel(logging.WARNING)
0021
0022 from django.conf import settings
0023 import httpx
0024 from monitor_app.teamcomms_auth import (
0025 PUBLIC_PREFIX, READ_SCOPES, WRITE_SCOPES, RemoteAuthentication, TrustedProxy,
0026 )
0027 from teamcomms.service.access import AccessError
0028
0029
0030 async def expect_denied(call, status):
0031 try:
0032 await call()
0033 except AccessError as error:
0034 assert error.status == status, (error.status, status)
0035 else:
0036 raise AssertionError("Request was admitted")
0037
0038
0039 async def main():
0040 payload = b'{"request":{}}'
0041 scope = {"type": "http", "method": "POST", "path": PUBLIC_PREFIX + "/mcp/",
0042 "query_string": b"one=a%20b", "headers": [(b"x-teamcomms-auth-ref", b"reference")],
0043 "teamcomms.body_sha256": hashlib.sha256(payload).hexdigest()}
0044 record = {"subject": "12", "username": "member", "name": "Member", "kind": "human",
0045 "auth_method": "session", "csrf_verified": True,
0046 "expires_at": (datetime.now(timezone.utc) + timedelta(seconds=60)).isoformat(),
0047 "method": "POST", "path": "/mcp/", "query_string": "one=a%20b",
0048 "body_sha256": hashlib.sha256(payload).hexdigest()}
0049 current = deepcopy(record)
0050 remote_status = 200
0051
0052 def respond(request):
0053 assert request.headers["authorization"] == "Bearer synthetic-service-key"
0054 return httpx.Response(remote_status, json=current)
0055
0056 with tempfile.NamedTemporaryFile(mode="w") as token_file:
0057 token_file.write("synthetic-service-key")
0058 token_file.flush()
0059 settings.SWF_TEAMCOMMS_SERVICE_TOKEN_FILE = token_file.name
0060 settings.SWF_TEAMCOMMS_INTROSPECTION_URL = "https://authority.invalid/introspect/"
0061 settings.SWF_TEAMCOMMS_PUBLIC_HOST = "epic-devcloud.org"
0062 auth = RemoteAuthentication()
0063 await auth.client.aclose()
0064 auth.client = httpx.AsyncClient(transport=httpx.MockTransport(respond))
0065 try:
0066 with patch("monitor_app.teamcomms_auth.database_call", AsyncMock(
0067 return_value=(READ_SCOPES | WRITE_SCOPES, "member"))):
0068 checked = deepcopy(scope)
0069 identity = await auth.resolve_request(checked)
0070 assert identity.subject == "12" and identity.session_authenticated
0071 assert await auth.check_request_csrf(checked, payload, identity)
0072 for key, value in [("method", "GET"), ("path", "/api/whoami"),
0073 ("query_string", "one=a+b"), ("body_sha256", "0" * 64)]:
0074 current = {**record, key: value}
0075 await expect_denied(lambda: auth.resolve_request(deepcopy(scope)), 401)
0076 current = {**record, "expires_at": "2020-01-01T00:00:00+00:00"}
0077 await expect_denied(lambda: auth.resolve_request(deepcopy(scope)), 401)
0078 current = {**record, "csrf_verified": False}
0079 checked = deepcopy(scope)
0080 identity = await auth.resolve_request(checked)
0081 assert not await auth.check_request_csrf(checked, payload, identity)
0082 current = {**record, "kind": "ai", "subject": "ai:12", "auth_method": "token",
0083 "operator": {"subject": "12", "username": "member", "name": "Member"}}
0084 identity = await auth.resolve_request(deepcopy(scope))
0085 assert identity.operator.subject == "12" and not identity.session_authenticated
0086 current["operator"]["username"] = "someone-else"
0087 await expect_denied(lambda: auth.resolve_request(deepcopy(scope)), 503)
0088 for kind in ("program", "connector"):
0089 service = {**record, "kind": kind, "subject": kind + ":12",
0090 "account_subject": "12", "auth_method": "token"}
0091 current = deepcopy(service)
0092 identity = await auth.resolve_request(deepcopy(scope))
0093 assert identity.kind == kind and identity.subject == kind + ":12"
0094 assert identity.operator is None and not identity.session_authenticated
0095 assert identity.scopes == READ_SCOPES | WRITE_SCOPES
0096 for key, value in [("auth_method", "session"), ("subject", kind + ":13"),
0097 ("subject", "ai:12"), ("account_subject", "13"),
0098 ("account_subject", "012"), ("account_subject", "１２"),
0099 ("account_subject", "0"), ("account_subject", 12),
0100 ("operator", {})]:
0101 current = {**service, key: value}
0102 await expect_denied(lambda: auth.resolve_request(deepcopy(scope)), 503)
0103 current = {key: value for key, value in service.items()
0104 if key != "account_subject"}
0105 await expect_denied(lambda: auth.resolve_request(deepcopy(scope)), 503)
0106 current = record
0107 for remote_status in (401, 403, 503):
0108 await expect_denied(lambda: auth.revalidate_request(deepcopy(scope), identity), remote_status)
0109 remote_status = 200
0110 duplicate = deepcopy(scope)
0111 duplicate["headers"].append((b"x-teamcomms-auth-ref", b"another"))
0112 await expect_denied(lambda: auth.resolve_request(duplicate), 401)
0113 finally:
0114 await auth.client.aclose()
0115
0116 captured = []
0117
0118 async def capture(incoming, receive, send):
0119 captured.append(incoming)
0120
0121 proxy = TrustedProxy(capture)
0122 incoming = {"type": "http", "client": ("127.0.0.1", 42), "path": "/swf-monitor/teamcomms/mcp/",
0123 "headers": [(b"x-teamcomms-auth-ref", b"reference"), (b"host", b"localhost"),
0124 (b"x-forwarded-host", b"epic-devcloud.org"), (b"x-forwarded-proto", b"https")]}
0125 await proxy(incoming, AsyncMock(), AsyncMock())
0126 assert captured[0]["path"] == "/prod/teamcomms/mcp/"
0127 assert captured[0]["scheme"] == "https"
0128 assert dict(captured[0]["headers"])[b"host"] == b"epic-devcloud.org"
0129 incoming["client"] = ("192.0.2.5", 42)
0130 sent = AsyncMock()
0131 await proxy(incoming, AsyncMock(), sent)
0132 assert sent.call_args_list[0].args[0]["status"] == 401 and len(captured) == 1
0133 print("TeamComms proxy binding, identity, CSRF and revalidation checks passed (no DB writes).")
0134
0135
0136 if __name__ == "__main__":
0137 asyncio.run(main())
