221 lines
7.7 KiB
Python
221 lines
7.7 KiB
Python
"""Regression contract for persistent ChatGPT OAuth connectivity."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
from urllib.request import Request
|
|
|
|
from scripts.promotion_preflight import _oidc_discovery_checks, run
|
|
from tests.test_promotion_preflight import _failed, _fixture
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class _DiscoveryResponse:
|
|
def __init__(
|
|
self, payload: dict[str, object], url: str, *, final_url: str | None = None
|
|
) -> None:
|
|
self.status = 200
|
|
self._body = json.dumps(payload).encode()
|
|
self._url = final_url or url
|
|
|
|
def __enter__(self) -> _DiscoveryResponse:
|
|
return self
|
|
|
|
def __exit__(self, *_: object) -> None:
|
|
return None
|
|
|
|
def read(self, _: int) -> bytes:
|
|
return self._body
|
|
|
|
def geturl(self) -> str:
|
|
return self._url
|
|
|
|
|
|
def _discovery_response(request: Any, *, refresh_capable: bool) -> _DiscoveryResponse:
|
|
url = str(request.full_url)
|
|
issuer = url.removesuffix("/.well-known/openid-configuration") + "/"
|
|
scopes = ["openid", "profile", "email"]
|
|
grants = ["authorization_code"]
|
|
if refresh_capable:
|
|
scopes.append("offline_access")
|
|
grants.append("refresh_token")
|
|
return _DiscoveryResponse(
|
|
{
|
|
"issuer": issuer,
|
|
"authorization_endpoint": "https://auth.test/application/o/authorize/",
|
|
"token_endpoint": "https://auth.test/application/o/token/",
|
|
"scopes_supported": scopes,
|
|
"grant_types_supported": grants,
|
|
"code_challenge_methods_supported": ["S256"],
|
|
},
|
|
url,
|
|
)
|
|
|
|
|
|
def test_authentik_scope_template_requests_offline_access() -> None:
|
|
env_example = (ROOT / "infrastructure/compose/.env.example").read_text(encoding="utf-8")
|
|
assert "AUTHENTIK_SCOPES=openid,profile,email,groups,offline_access" in env_example
|
|
|
|
|
|
def test_resource_metadata_and_challenge_share_refresh_capable_scopes() -> None:
|
|
from services.runtime.oauth_metadata import (
|
|
CHATGPT_OAUTH_SCOPES,
|
|
OAuthMetadataSettings,
|
|
bearer_challenge,
|
|
protected_resource_metadata,
|
|
)
|
|
|
|
expected = ("openid", "profile", "email", "offline_access")
|
|
settings = OAuthMetadataSettings(
|
|
public_origin="https://mcp.example.test",
|
|
issuer="https://auth.example.test/application/o/chatgpt/",
|
|
)
|
|
metadata = protected_resource_metadata("gitea-all/prod", settings)
|
|
challenge = bearer_challenge("gitea-all/prod", settings)
|
|
|
|
assert CHATGPT_OAUTH_SCOPES == expected
|
|
assert metadata["scopes_supported"] == list(expected)
|
|
assert metadata["authorization_servers"] == [settings.issuer]
|
|
assert metadata["resource"] == "https://mcp.example.test/mcp/gitea-all/prod"
|
|
assert f'scope="{" ".join(expected)}"' in challenge
|
|
assert "/.well-known/oauth-protected-resource/mcp/gitea-all/prod" in challenge
|
|
|
|
|
|
def test_oauth_metadata_settings_fail_closed_on_unsafe_values(
|
|
monkeypatch: Any,
|
|
) -> None:
|
|
from services.runtime.oauth_metadata import OAuthMetadataSettings
|
|
|
|
monkeypatch.setenv("HUB_PUBLIC_MCP_URL", "http://mcp.example.test")
|
|
monkeypatch.setenv(
|
|
"HUB_MCP_ADDITIONAL_OIDC_ISSUER", "https://auth.example.test/application/o/chatgpt/"
|
|
)
|
|
|
|
try:
|
|
OAuthMetadataSettings.from_environment()
|
|
except ValueError as exc:
|
|
assert "HUB_PUBLIC_MCP_URL" in str(exc)
|
|
else:
|
|
raise AssertionError("unsafe public OAuth origin was accepted")
|
|
|
|
|
|
def test_preflight_rejects_missing_offline_access(tmp_path: Path) -> None:
|
|
environment, _ = _fixture(tmp_path)
|
|
environment["AUTHENTIK_SCOPES"] = "openid,profile,email"
|
|
|
|
result = run(environment, "predeploy")
|
|
|
|
assert result["passed"] is False
|
|
assert _failed(result, "auth.coherence")
|
|
|
|
|
|
def test_preflight_accepts_refresh_capable_scope_contract(tmp_path: Path) -> None:
|
|
environment, _ = _fixture(tmp_path)
|
|
|
|
result = run(environment, "predeploy")
|
|
|
|
assert result["passed"] is True
|
|
assert not _failed(result, "auth.coherence")
|
|
|
|
|
|
def test_preflight_rejects_cross_host_chatgpt_issuer(tmp_path: Path) -> None:
|
|
environment, _ = _fixture(tmp_path)
|
|
environment["HUB_MCP_ADDITIONAL_OIDC_ISSUER"] = (
|
|
"https://other-auth.test/application/o/chatgpt-itworx-gitea/"
|
|
)
|
|
|
|
result = run(environment, "predeploy")
|
|
|
|
assert result["passed"] is False
|
|
assert _failed(result, "auth.coherence")
|
|
|
|
|
|
def test_preflight_rejects_non_strict_chatgpt_callback(tmp_path: Path) -> None:
|
|
environment, _ = _fixture(tmp_path)
|
|
environment["HUB_MCP_ADDITIONAL_OIDC_REDIRECT_URI"] = (
|
|
"https://chatgpt.com/connector/oauth/connector/extra"
|
|
)
|
|
|
|
result = run(environment, "predeploy")
|
|
|
|
assert result["passed"] is False
|
|
assert _failed(result, "auth.coherence")
|
|
|
|
|
|
def test_live_discovery_check_accepts_refresh_capable_providers() -> None:
|
|
environment = {
|
|
"HUB_OIDC_ISSUER": "https://auth.test/application/o/itworx/",
|
|
"HUB_MCP_ADDITIONAL_OIDC_ISSUER": ("https://auth.test/application/o/chatgpt-itworx-gitea/"),
|
|
}
|
|
results: list[dict[str, str]] = []
|
|
with patch(
|
|
"scripts.promotion_preflight._OIDC_DISCOVERY_OPENER.open",
|
|
side_effect=lambda request, timeout: _discovery_response(request, refresh_capable=True),
|
|
):
|
|
passed = _oidc_discovery_checks(environment, results, True)
|
|
|
|
assert passed is True
|
|
assert all(item["status"] == "pass" for item in results)
|
|
|
|
|
|
def test_live_discovery_check_rejects_missing_refresh_capability() -> None:
|
|
environment = {
|
|
"HUB_OIDC_ISSUER": "https://auth.test/application/o/itworx/",
|
|
"HUB_MCP_ADDITIONAL_OIDC_ISSUER": ("https://auth.test/application/o/chatgpt-itworx-gitea/"),
|
|
}
|
|
results: list[dict[str, str]] = []
|
|
with patch(
|
|
"scripts.promotion_preflight._OIDC_DISCOVERY_OPENER.open",
|
|
side_effect=lambda request, timeout: _discovery_response(request, refresh_capable=False),
|
|
):
|
|
passed = _oidc_discovery_checks(environment, results, True)
|
|
|
|
assert passed is False
|
|
assert all(item["status"] == "fail" for item in results)
|
|
|
|
|
|
def test_live_discovery_check_rejects_redirected_document() -> None:
|
|
environment = {
|
|
"HUB_OIDC_ISSUER": "https://auth.test/application/o/itworx/",
|
|
"HUB_MCP_ADDITIONAL_OIDC_ISSUER": ("https://auth.test/application/o/chatgpt-itworx-gitea/"),
|
|
}
|
|
results: list[dict[str, str]] = []
|
|
|
|
def redirected(request: Any, timeout: int) -> _DiscoveryResponse:
|
|
url = str(request.full_url)
|
|
issuer = url.removesuffix("/.well-known/openid-configuration") + "/"
|
|
return _DiscoveryResponse(
|
|
{
|
|
"issuer": issuer,
|
|
"authorization_endpoint": "https://auth.test/application/o/authorize/",
|
|
"token_endpoint": "https://auth.test/application/o/token/",
|
|
"scopes_supported": ["openid", "offline_access"],
|
|
"grant_types_supported": ["authorization_code", "refresh_token"],
|
|
"code_challenge_methods_supported": ["S256"],
|
|
},
|
|
url,
|
|
final_url="https://auth.test/unexpected-discovery",
|
|
)
|
|
|
|
with patch("scripts.promotion_preflight._OIDC_DISCOVERY_OPENER.open", side_effect=redirected):
|
|
passed = _oidc_discovery_checks(environment, results, True)
|
|
|
|
assert passed is False
|
|
assert all(item["status"] == "fail" for item in results)
|
|
|
|
|
|
def test_oidc_discovery_redirect_handler_never_follows_target() -> None:
|
|
from scripts.promotion_preflight import _RejectOidcDiscoveryRedirects
|
|
|
|
handler = _RejectOidcDiscoveryRedirects()
|
|
request = Request("https://auth.test/.well-known/openid-configuration")
|
|
|
|
assert (
|
|
handler.redirect_request(request, None, 302, "Found", {}, "http://127.0.0.1/internal")
|
|
is None
|
|
)
|