from __future__ import annotations import json import os import subprocess import sys import time from pathlib import Path import pytest from pydantic import ValidationError from sqlmodel import Session, SQLModel, create_engine, select FORGE_OPENID_MAPPING_ID = "75a34fd6-cc81-40b8-a8c3-2745af9567ab" FORGE_EMAIL_MAPPING_ID = "312810bb-11da-47d8-97d4-f2155348eaa2" FORGE_GROUP_MAPPING_ID = "7aa8c5f8-5e8e-4a1b-9c11-3a4f0a731cff" FORGE_APPLICATION_ID = "f7212687-a3d0-4eaa-b209-427f4043f870" FORGE_GROUP_IDS = { "forge-viewers": "01e9b457-75be-4c43-a276-5ef49eb691f7", "forge-operators": "6a238dab-1f85-4fce-b20d-c987a9d2924f", "forge-owners": "a3430246-574b-4e40-9217-59c12b7babca", } _OPENID_MAPPING_EXPRESSION = ( "# This scope is required by the OpenID-spec, and must as such exist in authentik.\n" "# The scope by itself does not grant any information\nreturn {}" ) _EMAIL_MAPPING_EXPRESSION = ( 'return {\n "email": request.user.email,\n "email_verified": False\n}' ) def _database(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): from backend.app.core.config import settings from backend.app.db import session as db_session from backend.app.db.init_db import init_db from backend.app.services import forge_oidc_provisioning # Windows CI temp roots may inherit shared-runner ACLs. These tests exercise # the provisioning contract; production's Windows ACL validator stays strict. if os.name == "nt": monkeypatch.setattr( forge_oidc_provisioning, "_windows_private_path", lambda *_args, **_kwargs: None, ) else: monkeypatch.setattr( forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", tmp_path / "forge-oidc-runtime-lock", ) engine = create_engine( f"sqlite:///{tmp_path / 'forge_oidc_test.db'}", connect_args={"check_same_thread": False}, ) monkeypatch.setattr(db_session, "engine", engine) monkeypatch.setattr(settings, "data_dir", str(tmp_path)) SQLModel.metadata.drop_all(engine) SQLModel.metadata.create_all(engine) init_db() return engine def _configured_settings(monkeypatch: pytest.MonkeyPatch): from backend.app.core.config import settings for key, value in { "AUTHENTIK_URL": "https://auth.example-host.com", "AUTHENTIK_TOKEN": "test-authentik-token-value", "AUTHENTIK_VERIFY_TLS": True, "AUTHENTIK_OIDC_AUTHORIZATION_FLOW": "authorization-flow-id", "AUTHENTIK_OIDC_INVALIDATION_FLOW": "invalidation-flow-id", "AUTHENTIK_OIDC_GROUP_CLAIM_MAPPING": "shared-generic-group-mapping-id", "AUTHENTIK_OIDC_ROLE_CLAIM": "groups", "AUTHENTIK_FORGE_OIDC_SIGNING_KEY_ID": "rsa-signing-key-id", "AUTHENTIK_FORGE_OIDC_BINDING_VALID": True, "AUTHENTIK_FORGE_OIDC_PROVIDER_ID": None, "AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING": FORGE_OPENID_MAPPING_ID, "AUTHENTIK_FORGE_OIDC_EMAIL_CLAIM_MAPPING": FORGE_EMAIL_MAPPING_ID, "AUTHENTIK_FORGE_OIDC_PROFILE_CLAIM_MAPPING": FORGE_GROUP_MAPPING_ID, }.items(): monkeypatch.setitem(settings.apis, key, value) monkeypatch.setattr(settings, "environment", "production") monkeypatch.setattr(settings, "secret_key", "test-production-secret-key-material-0123456789") monkeypatch.setattr(settings, "secret_key_source", "env") def _forge_group_mapping(expression: str | None = None): from backend.app.schemas.forge_oidc import ( FORGE_OIDC_PROFILE_MAPPING_EXPRESSION, FORGE_OIDC_PROFILE_MAPPING_NAME, ) return { "pk": FORGE_GROUP_MAPPING_ID, "name": FORGE_OIDC_PROFILE_MAPPING_NAME, "scope_name": "profile", "expression": expression or FORGE_OIDC_PROFILE_MAPPING_EXPRESSION, } def _valid_forge_group(name: str, group_id: str | None = None, *, users=None): return { "pk": group_id or FORGE_GROUP_IDS[name], "num_pk": {"forge-viewers": 14, "forge-operators": 15, "forge-owners": 16}[name], "name": name, "is_superuser": False, "attributes": {}, "parents": [], "children": [], "roles": [], "users": [] if users is None else users, } def _valid_scope_mappings(): from backend.app.schemas.forge_oidc import ( FORGE_OIDC_EMAIL_MAPPING_MANAGED, FORGE_OIDC_OPENID_MAPPING_MANAGED, ) return [ { "pk": FORGE_OPENID_MAPPING_ID, "managed": FORGE_OIDC_OPENID_MAPPING_MANAGED, "scope_name": "openid", "expression": _OPENID_MAPPING_EXPRESSION, }, { "pk": FORGE_EMAIL_MAPPING_ID, "managed": FORGE_OIDC_EMAIL_MAPPING_MANAGED, "scope_name": "email", "expression": _EMAIL_MAPPING_EXPRESSION, }, _forge_group_mapping(), ] def _valid_forge_groups(names=("forge-viewers", "forge-operators", "forge-owners")): return [_valid_forge_group(name) for name in names] def _mock_group_mapping_api(monkeypatch: pytest.MonkeyPatch, mapping=None): from backend.app.connectors.authentik import AuthentikConnector groups_by_id = {group["pk"]: group for group in _valid_forge_groups()} def request(self, path, params=None): if path.startswith("/core/groups/"): assert params == { "include_users": "true", "include_children": "true", "include_parents": "true", "include_inherited_roles": "false", } group_id = path.rstrip("/").split("/")[-1] return groups_by_id[group_id] if params is not None: raise AssertionError(f"Unexpected Authentik request: {path}") if path == f"/propertymappings/provider/scope/{FORGE_OPENID_MAPPING_ID}/": return { "pk": FORGE_OPENID_MAPPING_ID, "managed": "goauthentik.io/providers/oauth2/scope-openid", "scope_name": "openid", "expression": _OPENID_MAPPING_EXPRESSION, } if path == f"/propertymappings/provider/scope/{FORGE_EMAIL_MAPPING_ID}/": return { "pk": FORGE_EMAIL_MAPPING_ID, "managed": "goauthentik.io/providers/oauth2/scope-email", "scope_name": "email", "expression": _EMAIL_MAPPING_EXPRESSION, } if path == f"/propertymappings/provider/scope/{FORGE_GROUP_MAPPING_ID}/": return mapping if mapping is not None else _forge_group_mapping() raise AssertionError(f"Unexpected Authentik request: {path}") monkeypatch.setattr(AuthentikConnector, "_request", request, raising=True) def _install_forge_connector_remote( monkeypatch: pytest.MonkeyPatch, *, application_list=None, provider_list=None, provider_response_id: str = "31", provider_created_id: str = "31", provider_timeout_after_commit: bool = False, application_timeout_after_commit: bool = False, application_fail_without_commit: bool = False, ): from backend.app.connectors.authentik import AuthentikConnector _mock_group_mapping_api(monkeypatch) remote = { "applications": list(application_list or []), "providers": list(provider_list or []), } writes = [] behavior = { "provider_response_id": provider_response_id, "provider_created_id": provider_created_id, "provider_timeout_after_commit": provider_timeout_after_commit, "application_timeout_after_commit": application_timeout_after_commit, "application_fail_without_commit": application_fail_without_commit, } def paged(self, path, params=None): if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": assert params == {"has_key": True} return [{"pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True}] if path == "/core/groups/": return _valid_forge_groups() if path == "/propertymappings/provider/scope/": return _valid_scope_mappings() if path == "/core/applications/": return list(remote["applications"]) if path == "/providers/oauth2/": return list(remote["providers"]) raise AssertionError(f"Unexpected Authentik read: {path}") def write(self, method, path, payload=None): writes.append((method, path, payload)) if path == "/providers/oauth2/": provider = {"pk": behavior["provider_created_id"], **payload} remote["providers"].append(provider) if behavior["provider_timeout_after_commit"]: raise RuntimeError("request timeout") return {"pk": behavior["provider_response_id"]} if path == "/core/applications/": if behavior["application_fail_without_commit"]: raise RuntimeError("request timeout") remote["applications"].append({"pk": FORGE_APPLICATION_ID, **payload}) if behavior["application_timeout_after_commit"]: raise RuntimeError("request timeout") return {"pk": FORGE_APPLICATION_ID} raise AssertionError(f"Unexpected Authentik write: {method} {path}") monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_write_request", write, raising=False) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_provider_secret_matches", lambda *_args: True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_discovery_matches", lambda self: True) return remote, writes, behavior def _add_policy(session: Session, *, target: str = "itworx-forge", manual: str = "allowed"): from backend.app.db.models import Policy existing = session.exec( select(Policy).where( (Policy.action == "forge.oidc.provision") & (Policy.target == target) ) ).first() if existing is not None: existing.manual = manual session.add(existing) session.commit() session.refresh(existing) return existing policy = Policy( id=f"forge-oidc-policy-{target}", action="forge.oidc.provision", target=target, voice="blocked", manual=manual, risk="critical", reason="Provision the fixed Forge OIDC client after explicit operator approval.", ) session.add(policy) session.commit() return policy class RecordingAuthentik: def __init__( self, result: dict[str, object] | None = None, preflight: dict[str, object] | None = None, remote_client_secret: str | None = None, ): self.calls: list[str] = [] self.preflight_calls = 0 self.verified_secrets: list[str] = [] self.preflight = preflight or {"success": True, "status": "ready"} self.remote_client_secret = remote_client_secret self.result = result or { "success": True, "status": "created", "message": "forge_oidc_created", "applicationId": "f7212687-a3d0-4eaa-b209-427f4043f870", "providerId": "31", } def inspect_forge_oidc_application(self, *, expected_client_secret: str | None = None): self.preflight_calls += 1 if expected_client_secret is not None and self.preflight.get("status") in { "already_configured", "provider_orphan", }: self.verified_secrets.append(expected_client_secret) if expected_client_secret != self.remote_client_secret: return { "success": False, "status": "conflict", "message": "forge_oidc_existing_client_secret_mismatch", } return self.preflight def ensure_forge_oidc_prerequisites(self, *, expected_client_secret: str | None = None): return { "success": True, "status": "ready", "scopeMappingIds": { "openid": FORGE_OPENID_MAPPING_ID, "email": FORGE_EMAIL_MAPPING_ID, "profile": FORGE_GROUP_MAPPING_ID, }, } def ensure_forge_oidc_application(self, *, client_secret: str, persist_provider_id): verified = self.inspect_forge_oidc_application(expected_client_secret=client_secret) if verified.get("success") is not True: return verified if verified.get("status") != "already_configured": self.calls.append(client_secret) provider_id = str(verified.get("providerId") or self.result.get("providerId") or "") if not persist_provider_id(provider_id): return {"success": False, "status": "partial_failure", "message": "forge_oidc_binding_persist_failed"} if self.preflight.get("status") == "already_configured": return verified return self.result def test_forge_oidc_request_schema_rejects_caller_selected_targets(): from backend.app.schemas.forge_oidc import ForgeOidcProvisionRequestV1 assert ForgeOidcProvisionRequestV1().model_dump() == {} with pytest.raises(ValidationError): ForgeOidcProvisionRequestV1.model_validate({"client_id": "other-client"}) def test_forge_private_package_owner_has_no_default_and_reports_presence_only( monkeypatch: pytest.MonkeyPatch, ): from backend.app.core.config import Settings monkeypatch.delenv("FORGE_PRIVATE_PACKAGE_OWNER", raising=False) unset = Settings() assert unset.apis["FORGE_PRIVATE_PACKAGE_OWNER"] is None assert unset.redact_env_preview()["FORGE_PRIVATE_PACKAGE_OWNER"] is False monkeypatch.setenv("FORGE_PRIVATE_PACKAGE_OWNER", "trusted-forge-owner") configured = Settings() assert configured.apis["FORGE_PRIVATE_PACKAGE_OWNER"] == "trusted-forge-owner" assert configured.redact_env_preview()["FORGE_PRIVATE_PACKAGE_OWNER"] is True def test_forge_oidc_profile_mapping_emits_only_allowlisted_groups_from_ak_groups(): from types import SimpleNamespace from backend.app.schemas.forge_oidc import FORGE_OIDC_PROFILE_MAPPING_EXPRESSION source = "def evaluate(request, delete_none_values):\n" + "\n".join( f" {line}" if line else "" for line in FORGE_OIDC_PROFILE_MAPPING_EXPRESSION.splitlines() ) namespace: dict[str, object] = {} exec(compile(source, "", "exec"), namespace) user = SimpleNamespace( name="Forge User", username="forge-user", ak_groups=SimpleNamespace( all=lambda: [ SimpleNamespace(name="forge-owners"), SimpleNamespace(name="unrelated-admin"), SimpleNamespace(name="forge-viewers"), ] ), ) result = namespace["evaluate"]( # type: ignore[operator] SimpleNamespace(user=user), lambda claims: {k: v for k, v in claims.items() if v is not None} ) assert result == { "name": "Forge User", "preferred_username": "forge-user", "groups": ["forge-owners", "forge-viewers"], } assert not hasattr(user, "groups") def test_forge_oidc_binding_schema_rejects_duplicate_and_caller_selected_ids(): from backend.app.schemas.forge_oidc import ( ForgeOidcBindingV1, ForgeOidcScopeMappingIdsV1, ) shared = "75a34fd6-cc81-40b8-a8c3-2745af9567ab" with pytest.raises(ValidationError): ForgeOidcBindingV1.model_validate( { "version": 1, "providerId": None, "scopeMappingIds": {"openid": shared, "email": shared, "profile": shared}, } ) base = { "providerId": None, "scopeMappingIds": { "openid": FORGE_OPENID_MAPPING_ID, "email": FORGE_EMAIL_MAPPING_ID, "profile": FORGE_GROUP_MAPPING_ID, }, } for invalid_version in (True, 1.0, "1", 2): with pytest.raises(ValidationError): ForgeOidcBindingV1.model_validate({**base, "version": invalid_version}) for invalid_provider_id in ("0", "01", "12345678901", "١٢٣", " 31"): with pytest.raises(ValidationError): ForgeOidcBindingV1.model_validate( {**base, "version": 1, "providerId": invalid_provider_id} ) for invalid_uuid in ( "00000000-0000-0000-0000-000000000000", FORGE_OPENID_MAPPING_ID.upper(), "not-a-uuid", ): with pytest.raises(ValidationError): ForgeOidcScopeMappingIdsV1.model_validate( { "openid": invalid_uuid, "email": FORGE_EMAIL_MAPPING_ID, "profile": FORGE_GROUP_MAPPING_ID, } ) with pytest.raises(ValidationError): ForgeOidcBindingV1.model_validate( { "version": 1, "providerId": None, "scopeMappingIds": { "openid": shared, "email": "312810bb-11da-47d8-97d4-f2155348eaa2", "profile": "de878051-a898-42ce-b46f-bd7d9b1dd171", "groups": shared, }, } ) def test_forge_oidc_binding_loader_is_atomic_and_fails_closed_on_tampering( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): import json from backend.app.core.config import settings from backend.app.schemas.forge_oidc import FORGE_OIDC_BINDING_SECRET_REF from backend.app.services.secret_store import set_secret from backend.app.services.setup_service import _apply_forge_oidc_binding_to_runtime engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) valid = { "version": 1, "providerId": "31", "scopeMappingIds": { "openid": "75a34fd6-cc81-40b8-a8c3-2745af9567ab", "email": "312810bb-11da-47d8-97d4-f2155348eaa2", "profile": FORGE_GROUP_MAPPING_ID, }, } with Session(engine) as session: set_secret(session, FORGE_OIDC_BINDING_SECRET_REF, json.dumps(valid)) assert _apply_forge_oidc_binding_to_runtime(session) is True assert settings.apis["AUTHENTIK_FORGE_OIDC_BINDING_VALID"] is True assert settings.apis["AUTHENTIK_FORGE_OIDC_PROVIDER_ID"] == "31" assert settings.apis["AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING"] == valid["scopeMappingIds"]["openid"] assert json.dumps(valid) not in str(settings.apis) tampered = dict(valid) tampered["scopeMappingIds"] = { "openid": valid["scopeMappingIds"]["openid"], "email": valid["scopeMappingIds"]["openid"], "profile": valid["scopeMappingIds"]["profile"], } set_secret(session, FORGE_OIDC_BINDING_SECRET_REF, json.dumps(tampered)) assert _apply_forge_oidc_binding_to_runtime(session) is False duplicate_key = ( '{"version":1,"version":1,"providerId":"31",' '"scopeMappingIds":{"openid":"75a34fd6-cc81-40b8-a8c3-2745af9567ab",' '"email":"312810bb-11da-47d8-97d4-f2155348eaa2",' f'"profile":"{FORGE_GROUP_MAPPING_ID}"}}' ) set_secret(session, FORGE_OIDC_BINDING_SECRET_REF, duplicate_key) assert _apply_forge_oidc_binding_to_runtime(session) is False assert settings.apis["AUTHENTIK_FORGE_OIDC_BINDING_VALID"] is False assert settings.apis["AUTHENTIK_FORGE_OIDC_PROVIDER_ID"] is None assert all(settings.apis[key] is None for key in ( "AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING", "AUTHENTIK_FORGE_OIDC_EMAIL_CLAIM_MAPPING", "AUTHENTIK_FORGE_OIDC_PROFILE_CLAIM_MAPPING", )) def test_forge_oidc_provisioning_stays_blocked_without_registered_exact_policy( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.db.models import AuditEvent from backend.app.services.forge_oidc_provisioning import provision_forge_oidc engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() with Session(engine) as session: from backend.app.db.models import Policy exact = session.exec( select(Policy).where( (Policy.action == "forge.oidc.provision") & (Policy.target == "itworx-forge") ) ).one() session.delete(exact) session.commit() result = provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) events = session.exec(select(AuditEvent)).all() assert result["status"] == "blocked" assert result["message"] == "forge_oidc_exact_target_policy_required" assert authentik.calls == [] assert events[0].action_id == "forge.oidc.provision" assert events[0].target == "itworx-forge" assert events[0].policy_decision == "blocked" def test_forge_oidc_registered_action_requires_exact_manual_policy( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.services.forge_oidc_provisioning import provision_forge_oidc engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() with Session(engine) as session: _add_policy(session) result = provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) assert result["status"] == "created" assert authentik.preflight_calls == 2 assert len(authentik.calls) == 1 def test_forge_oidc_wildcard_policy_cannot_authorize_mutation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.db.models import AuditEvent from backend.app.services import forge_oidc_provisioning engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() with Session(engine) as session: _add_policy(session, target="*") result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) event = session.exec(select(AuditEvent)).one() assert result["status"] == "blocked" assert result["message"] == "forge_oidc_exact_target_policy_required" assert authentik.calls == [] assert event.policy_decision == "blocked" def test_forge_oidc_policy_is_one_exact_manual_critical_row_and_voice_blocked( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.db.models import Policy from backend.app.services.action_registry import ( ACTION_RISK, ACTIONS, action_default_target, ) from backend.app.services.policy_engine import evaluate_policy engine = _database(tmp_path, monkeypatch) with Session(engine) as session: policy = session.exec( select(Policy).where(Policy.action == "forge.oidc.provision") ).one() assert "forge.oidc.provision" in ACTIONS assert ACTION_RISK["forge.oidc.provision"] == "critical" assert action_default_target("forge.oidc.provision") == "itworx-forge" assert (policy.target, policy.manual, policy.voice, policy.risk) == ( "itworx-forge", "allowed", "blocked", "critical", ) assert evaluate_policy( session, action="forge.oidc.provision", target="itworx-forge", source="manual", )["decision"] == "allowed" for source, target in [("voice", "itworx-forge"), ("manual", "other-app")]: assert evaluate_policy( session, action="forge.oidc.provision", target=target, source=source, )["decision"] == "blocked" assert evaluate_policy( session, action="authentik.create_provider", target="authentik", source="manual", )["decision"] == "blocked" session.add( Policy( id="policy-forge-oidc-wildcard-test", action="forge.oidc.provision", target="*", voice="blocked", manual="allowed", risk="critical", reason="Must never authorize Forge provisioning.", ) ) session.commit() assert evaluate_policy( session, action="forge.oidc.provision", target="itworx-forge", source="manual", )["decision"] == "blocked" session.delete( session.exec( select(Policy).where(Policy.id == "policy-forge-oidc-wildcard-test") ).one() ) session.add( Policy( id="policy-forge-oidc-duplicate-exact-test", action="forge.oidc.provision", target="itworx-forge", voice="blocked", manual="allowed", risk="critical", reason="A second exact row must fail closed.", ) ) session.commit() assert evaluate_policy( session, action="forge.oidc.provision", target="itworx-forge", source="manual", )["decision"] == "blocked" def test_forge_oidc_provisioning_uses_exact_policy_stores_secret_and_returns_no_secret( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.db.models import AuditEvent from backend.app.services import forge_oidc_provisioning from backend.app.services.secret_store import get_secret, has_secret engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() generated = [] monkeypatch.setattr( forge_oidc_provisioning, "_new_client_secret", lambda: generated.append("one-time-client-secret-value-0123456789") or "one-time-client-secret-value-0123456789", ) with Session(engine) as session: policy = _add_policy(session) monkeypatch.setattr( forge_oidc_provisioning, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, }, ) result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) stored_secret = get_secret(session, "ITWORX_FORGE_OIDC_CLIENT_SECRET", touch=False) events = session.exec(select(AuditEvent)).all() assert result["status"] == "created" assert result["clientId"] == "itworx-forge" assert result["callbackUrl"] == "https://forge.example.com/auth/callback" assert result["issuer"] == "https://auth.example-host.com/application/o/itworx-forge/" assert result["groups"] == ["forge-viewers", "forge-operators", "forge-owners"] assert result["secretRef"] == "ITWORX_FORGE_OIDC_CLIENT_SECRET" assert "one-time-client-secret-value-0123456789" not in str(result) assert authentik.calls == ["one-time-client-secret-value-0123456789"] assert stored_secret == "one-time-client-secret-value-0123456789" assert has_secret(session, "ITWORX_FORGE_OIDC_CLIENT_SECRET") binding_json = get_secret(session, "AUTHENTIK_FORGE_OIDC_BINDING", touch=False) assert binding_json is not None assert json.loads(binding_json) == { "version": 1, "providerId": "31", "scopeMappingIds": { "openid": FORGE_OPENID_MAPPING_ID, "email": FORGE_EMAIL_MAPPING_ID, "profile": FORGE_GROUP_MAPPING_ID, }, } assert "one-time-client-secret-value-0123456789" not in str(events) assert generated == ["one-time-client-secret-value-0123456789"] def test_forge_oidc_provisioning_reuses_existing_secret_on_retry( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.services import forge_oidc_provisioning from backend.app.services.secret_store import get_secret, set_secret engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() generated = [] monkeypatch.setattr( forge_oidc_provisioning, "_new_client_secret", lambda: generated.append("unexpected") or "unexpected", ) with Session(engine) as session: policy = _add_policy(session) set_secret( session, "ITWORX_FORGE_OIDC_CLIENT_SECRET", "persisted-client-secret-value-0123456789", used_by=["authentik:itworx-forge"], ) monkeypatch.setattr( forge_oidc_provisioning, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, }, ) result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) stored_secret = get_secret(session, "ITWORX_FORGE_OIDC_CLIENT_SECRET", touch=False) assert result["status"] == "created" assert authentik.calls == ["persisted-client-secret-value-0123456789"] assert stored_secret == "persisted-client-secret-value-0123456789" assert generated == [] def test_forge_oidc_existing_remote_client_requires_local_secret_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.services import forge_oidc_provisioning engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik( preflight={ "success": True, "status": "already_configured", "applicationId": "f7212687-a3d0-4eaa-b209-427f4043f870", "providerId": "31", } ) with Session(engine) as session: policy = _add_policy(session) monkeypatch.setattr( forge_oidc_provisioning, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, }, ) result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) assert result["status"] == "conflict" assert result["message"] == "forge_oidc_existing_client_secret_missing" assert authentik.preflight_calls == 1 assert authentik.calls == [] def test_forge_oidc_existing_remote_client_with_local_secret_is_idempotent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.services import forge_oidc_provisioning from backend.app.services.secret_store import set_secret engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik( preflight={ "success": True, "status": "already_configured", "applicationId": "f7212687-a3d0-4eaa-b209-427f4043f870", "providerId": "31", }, remote_client_secret="persisted-client-secret-value-0123456789", ) with Session(engine) as session: policy = _add_policy(session) set_secret( session, "ITWORX_FORGE_OIDC_CLIENT_SECRET", "persisted-client-secret-value-0123456789", used_by=["authentik:itworx-forge"], ) monkeypatch.setattr( forge_oidc_provisioning, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, }, ) result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) assert result["status"] == "already_configured" assert result["secretCreated"] is False assert authentik.preflight_calls == 2 assert authentik.verified_secrets == [ "persisted-client-secret-value-0123456789", "persisted-client-secret-value-0123456789", ] assert authentik.calls == [] def test_forge_oidc_existing_remote_client_secret_mismatch_is_conflict_and_redacted( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.services import forge_oidc_provisioning from backend.app.services.secret_store import set_secret engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) local_secret = "persisted-local-client-secret-value" remote_secret = "rotated-remote-client-secret-value" authentik = RecordingAuthentik( preflight={ "success": True, "status": "already_configured", "applicationId": "f7212687-a3d0-4eaa-b209-427f4043f870", "providerId": "31", }, remote_client_secret=remote_secret, ) with Session(engine) as session: policy = _add_policy(session) set_secret(session, "ITWORX_FORGE_OIDC_CLIENT_SECRET", local_secret) monkeypatch.setattr( forge_oidc_provisioning, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, }, ) result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) assert result["status"] == "conflict" assert result["message"] == "forge_oidc_resource_conflict" assert local_secret not in str(result) assert remote_secret not in str(result) assert authentik.verified_secrets == [local_secret] assert authentik.calls == [] def test_forge_oidc_concurrent_calls_create_one_pair_and_keep_one_secret( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ): from concurrent.futures import ThreadPoolExecutor from threading import RLock from backend.app.services import forge_oidc_provisioning from backend.app.services.secret_store import get_secret engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) policy_lock = RLock() class ConcurrentAuthentik(RecordingAuthentik): def __init__(self): super().__init__() self.remote_lock = RLock() self.remote_created = False self.remote_secret = None self.create_count = 0 def inspect_forge_oidc_application(self, *, expected_client_secret=None): with self.remote_lock: if not self.remote_created: return {"success": True, "status": "ready", "message": "forge_oidc_preflight_ready"} if expected_client_secret is not None and expected_client_secret != self.remote_secret: return {"success": False, "status": "conflict", "message": "forge_oidc_existing_client_secret_mismatch"} return { "success": True, "status": "already_configured", "message": "forge_oidc_already_configured", "applicationId": FORGE_APPLICATION_ID, "providerId": "31", } def ensure_forge_oidc_application(self, *, client_secret: str, persist_provider_id): with self.remote_lock: observed = self.inspect_forge_oidc_application(expected_client_secret=client_secret) if observed.get("success") is not True: return observed if observed.get("status") == "already_configured": if not persist_provider_id("31"): return {"success": False, "status": "partial_failure", "message": "forge_oidc_binding_persist_failed"} return observed if not persist_provider_id("31"): return {"success": False, "status": "partial_failure", "message": "forge_oidc_binding_persist_failed"} self.remote_created = True self.remote_secret = client_secret self.create_count += 1 self.calls.append(client_secret) return { "success": True, "status": "created", "message": "forge_oidc_created", "applicationId": FORGE_APPLICATION_ID, "providerId": "31", } authentik = ConcurrentAuthentik() with Session(engine) as session: policy = _add_policy(session) def allow_policy(*_args, **_kwargs): with policy_lock: return { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, } monkeypatch.setattr(forge_oidc_provisioning, "evaluate_policy", allow_policy) def provision_once(): with Session(engine) as session: return forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) with ThreadPoolExecutor(max_workers=2) as pool: results = list(pool.map(lambda _index: provision_once(), range(2))) assert sorted(result["status"] for result in results) == ["already_configured", "created"] assert authentik.create_count == 1 assert len(authentik.calls) == 1 with Session(engine) as session: stored = get_secret(session, "ITWORX_FORGE_OIDC_CLIENT_SECRET", touch=False) binding = json.loads(get_secret(session, "AUTHENTIK_FORGE_OIDC_BINDING", touch=False)) assert stored == authentik.remote_secret assert binding["providerId"] == "31" def test_forge_oidc_provisioning_blocks_without_production_secret_storage( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.core.config import settings from backend.app.services import forge_oidc_provisioning engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() monkeypatch.setattr(settings, "secret_key_source", "missing") monkeypatch.setattr(settings, "secret_key", None) with Session(engine) as session: policy = _add_policy(session) monkeypatch.setattr( forge_oidc_provisioning, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": policy.id, "risk": "critical", "reason": policy.reason, }, ) result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) assert result["status"] == "blocked" assert result["message"] == "forge_oidc_secret_store_not_production_ready" assert authentik.calls == [] def test_authentik_connector_creates_only_the_fixed_forge_oidc_contract( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) writes = [] remote = {"applications": [], "providers": []} def paged(self, path, params=None): if path == "/core/groups/": return _valid_forge_groups() if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": assert params == {"has_key": True} return [ { "pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True, } ] if path == "/propertymappings/provider/scope/": return _valid_scope_mappings() if path in {"/core/applications/", "/providers/oauth2/"}: key = "applications" if path == "/core/applications/" else "providers" return list(remote[key]) raise AssertionError(f"Unexpected Authentik read: {path}") def write(self, method, path, payload=None): writes.append((method, path, payload)) if path == "/providers/oauth2/": remote["providers"].append( {"pk": "31", **payload, "property_mappings": list(reversed(payload["property_mappings"]))} ) return {"pk": "31"} if path == "/core/applications/": remote["applications"].append( {"pk": FORGE_APPLICATION_ID, **payload} ) return {"pk": "f7212687-a3d0-4eaa-b209-427f4043f870"} raise AssertionError(f"Unexpected Authentik write: {method} {path}") monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_write_request", write, raising=False) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_provider_secret_matches", lambda *_args: True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_discovery_matches", lambda self: True) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="one-time-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: True, ) observed = AuthentikConnector().inspect_forge_oidc_application( expected_client_secret="one-time-client-secret-value-0123456789" ) assert observed["success"] is True, observed["message"] assert result["success"] is True, result["message"] assert result["status"] == "created" assert result["applicationId"] == "f7212687-a3d0-4eaa-b209-427f4043f870" assert result["providerId"] == "31" assert writes[0][0:2] == ("POST", "/providers/oauth2/") assert writes[0][2] == { "name": "ITWorx Forge OIDC", "authorization_flow": "authorization-flow-id", "invalidation_flow": "invalidation-flow-id", "client_type": "confidential", "client_id": "itworx-forge", "client_secret": "one-time-client-secret-value-0123456789", "signing_key": "rsa-signing-key-id", "redirect_uris": [{"matching_mode": "strict", "url": "https://forge.example.com/auth/callback"}], "property_mappings": [FORGE_OPENID_MAPPING_ID, FORGE_EMAIL_MAPPING_ID, FORGE_GROUP_MAPPING_ID], "include_claims_in_id_token": True, "sub_mode": "hashed_user_id", "issuer_mode": "per_provider", } assert writes[1] == ( "POST", "/core/applications/", { "name": "ITWorx Forge", "slug": "itworx-forge", "provider": "31", "meta_launch_url": "https://forge.example.com", "meta_description": "Managed by AppOps Forge OIDC v1.", }, ) assert "one-time-client-secret-value-0123456789" not in str(result) def test_authentik_connector_accepts_exact_existing_contract_without_writes( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) writes = [] def paged(self, path, params=None): if path == "/core/groups/": return _valid_forge_groups() if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [ { "pk": "rsa-signing-key-id", "key_type": "RSA", "private_key_available": True, } ] if path == "/propertymappings/provider/scope/": return _valid_scope_mappings() if path == "/core/applications/": return [ { "pk": FORGE_APPLICATION_ID, "name": "ITWorx Forge", "slug": "itworx-forge", "provider": "31", "meta_launch_url": "https://forge.example.com", "meta_description": "Managed by AppOps Forge OIDC v1.", } ] if path == "/providers/oauth2/": return [ { "pk": "31", "name": "ITWorx Forge OIDC", "client_id": "itworx-forge", "client_type": "confidential", "redirect_uris": [ { "matching_mode": "strict", "url": "https://forge.example.com/auth/callback", } ], "authorization_flow": "authorization-flow-id", "invalidation_flow": "invalidation-flow-id", "property_mappings": [FORGE_OPENID_MAPPING_ID, FORGE_EMAIL_MAPPING_ID, FORGE_GROUP_MAPPING_ID], "include_claims_in_id_token": True, "sub_mode": "hashed_user_id", "issuer_mode": "per_provider", "signing_key": "rsa-signing-key-id", } ] raise AssertionError(f"Unexpected Authentik read: {path}") monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_discovery_matches", lambda self: True) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_provider_secret_matches", lambda self, provider_id, expected: provider_id == "31" and expected == "unused-client-secret-value-0123456789", raising=True, ) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_write_request", lambda self, method, path, payload=None: writes.append((method, path, payload)), raising=False, ) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="unused-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: True, ) assert result == { "success": True, "status": "already_configured", "message": "forge_oidc_already_configured", "applicationId": "f7212687-a3d0-4eaa-b209-427f4043f870", "providerId": "31", } assert writes == [] def test_authentik_provider_contract_uses_mapping_identity_not_relation_order( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) provider = { "pk": "31", "name": "ITWorx Forge OIDC", "client_id": "itworx-forge", "client_type": "confidential", "redirect_uris": [ {"matching_mode": "strict", "url": "https://forge.example.com/auth/callback"} ], "authorization_flow": "authorization-flow-id", "invalidation_flow": "invalidation-flow-id", "property_mappings": [ FORGE_GROUP_MAPPING_ID, FORGE_EMAIL_MAPPING_ID, FORGE_OPENID_MAPPING_ID, ], "include_claims_in_id_token": True, "sub_mode": "hashed_user_id", "issuer_mode": "per_provider", "signing_key": "rsa-signing-key-id", } assert AuthentikConnector._forge_oidc_provider_matches( provider, signing_key_id="rsa-signing-key-id", ) is True for unsupported_field, value in (("grant_types", ["authorization_code"]), ("redirect_uri_type", "strict")): assert AuthentikConnector._forge_oidc_provider_matches( {**provider, unsupported_field: value}, signing_key_id="rsa-signing-key-id", ) is False def test_authentik_group_numeric_alternate_id_is_never_used_as_group_pk( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) numeric_group = _valid_forge_group("forge-viewers", "42") numeric_group["num_pk"] = 42 def paged(self, path, params=None): if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [{"pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True}] if path == "/core/groups/": return [numeric_group] if path == "/propertymappings/provider/scope/": return _valid_scope_mappings() return [] writes = [] monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_write_request", lambda self, method, path, payload=None: writes.append((method, path, payload)), raising=False, ) result = AuthentikConnector().ensure_forge_oidc_prerequisites() assert result["status"] == "conflict" assert result["message"] == "forge_oidc_group_id_invalid" assert writes == [] def test_authentik_group_membership_conflict_blocks_prerequisite_mutations( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) def paged(self, path, params=None): if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [{"pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True}] if path == "/core/groups/": return _valid_forge_groups() if path == "/propertymappings/provider/scope/": return _valid_scope_mappings() return [] def read_group(self, group_id): group = next(item for item in _valid_forge_groups() if item["pk"] == group_id) if group_id == FORGE_GROUP_IDS["forge-viewers"]: return {**group, "users": ["unexpected-user"]} return group writes = [] monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_read_group", read_group, raising=True) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_write_request", lambda self, method, path, payload=None: writes.append((method, path, payload)), raising=False, ) result = AuthentikConnector().ensure_forge_oidc_prerequisites() assert result["status"] == "conflict" assert result["message"] == "forge_oidc_group_conflict" assert writes == [] def test_authentik_bound_exact_application_preserves_existing_group_memberships( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector from backend.app.core.config import settings _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) monkeypatch.setitem(settings.apis, "AUTHENTIK_FORGE_OIDC_PROVIDER_ID", "31") secret = "stored-client-secret-value-0123456789" provider = { "pk": "31", "name": "ITWorx Forge OIDC", "client_id": "itworx-forge", "client_type": "confidential", "redirect_uris": [ {"matching_mode": "strict", "url": "https://forge.example.com/auth/callback"} ], "authorization_flow": "authorization-flow-id", "invalidation_flow": "invalidation-flow-id", "property_mappings": [FORGE_OPENID_MAPPING_ID, FORGE_EMAIL_MAPPING_ID, FORGE_GROUP_MAPPING_ID], "include_claims_in_id_token": True, "sub_mode": "hashed_user_id", "issuer_mode": "per_provider", "signing_key": "rsa-signing-key-id", } application = { "pk": FORGE_APPLICATION_ID, "name": "ITWorx Forge", "slug": "itworx-forge", "provider": "31", "meta_launch_url": "https://forge.example.com", "meta_description": "Managed by AppOps Forge OIDC v1.", } remote, writes, _behavior = _install_forge_connector_remote( monkeypatch, application_list=[application], provider_list=[provider], ) def read_group(self, group_id): name = next(name for name, value in FORGE_GROUP_IDS.items() if value == group_id) users = ["existing-forge-user"] if name == "forge-viewers" else [] return _valid_forge_group(name, users=users) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_read_group", read_group, raising=True) result = AuthentikConnector().ensure_forge_oidc_prerequisites(expected_client_secret=secret) assert result["success"] is True assert result["status"] == "ready" assert writes == [] assert len(remote["applications"]) == len(remote["providers"]) == 1 def test_authentik_prerequisites_create_only_absent_groups_and_reconcile_unknown_post( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector from backend.app.core.config import settings _configured_settings(monkeypatch) monkeypatch.setitem(settings.apis, "AUTHENTIK_FORGE_OIDC_BINDING_VALID", False) for key in ( "AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING", "AUTHENTIK_FORGE_OIDC_EMAIL_CLAIM_MAPPING", "AUTHENTIK_FORGE_OIDC_PROFILE_CLAIM_MAPPING", ): monkeypatch.setitem(settings.apis, key, None) from backend.app.schemas.forge_oidc import ( FORGE_OIDC_EMAIL_MAPPING_MANAGED, FORGE_OIDC_OPENID_MAPPING_MANAGED, FORGE_OIDC_PROFILE_MAPPING_EXPRESSION, FORGE_OIDC_PROFILE_MAPPING_NAME, ) state = { "groups": [], "mappings": [ { "pk": FORGE_OPENID_MAPPING_ID, "managed": FORGE_OIDC_OPENID_MAPPING_MANAGED, "scope_name": "openid", "expression": _OPENID_MAPPING_EXPRESSION, }, { "pk": FORGE_EMAIL_MAPPING_ID, "managed": FORGE_OIDC_EMAIL_MAPPING_MANAGED, "scope_name": "email", "expression": _EMAIL_MAPPING_EXPRESSION, }, ], "group_details": {}, "mapping_details": {}, "unknown_group_post": True, } writes = [] def paged(self, path, params=None): if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [{"pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True}] if path == "/core/groups/": return list(state["groups"]) if path == "/propertymappings/provider/scope/": return list(state["mappings"]) raise AssertionError(f"Unexpected Authentik read: {path}") def request(self, path, params=None): if path.startswith("/core/groups/"): assert params == { "include_users": "true", "include_children": "true", "include_parents": "true", "include_inherited_roles": "false", } return state["group_details"][path.rstrip("/").split("/")[-1]] if path.startswith("/propertymappings/provider/scope/"): return state["mapping_details"][path.rstrip("/").split("/")[-1]] raise AssertionError(f"Unexpected Authentik request: {path}") def write(self, method, path, payload=None): writes.append((method, path, payload)) if path == "/core/groups/": name = payload["name"] group_id = FORGE_GROUP_IDS[name] group = {"pk": group_id, **payload, "num_pk": 100 + len(state["groups"])} state["groups"].append(group) state["group_details"][group_id] = group if state["unknown_group_post"]: state["unknown_group_post"] = False raise RuntimeError("timeout after commit") return {"pk": group_id} if path == "/propertymappings/provider/scope/": mapping = {"pk": FORGE_GROUP_MAPPING_ID, **payload} state["mappings"].append(mapping) state["mapping_details"][FORGE_GROUP_MAPPING_ID] = mapping return {"pk": FORGE_GROUP_MAPPING_ID} raise AssertionError(f"Unexpected Authentik write: {method} {path}") monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr(AuthentikConnector, "_request", request, raising=True) monkeypatch.setattr(AuthentikConnector, "_forge_oidc_write_request", write, raising=False) result = AuthentikConnector().ensure_forge_oidc_prerequisites() assert result["success"] is True assert result["scopeMappingIds"] == { "openid": FORGE_OPENID_MAPPING_ID, "email": FORGE_EMAIL_MAPPING_ID, "profile": FORGE_GROUP_MAPPING_ID, } assert [path for _method, path, _payload in writes] == [ "/core/groups/", "/core/groups/", "/core/groups/", "/propertymappings/provider/scope/", ] assert all(group["is_superuser"] is False and group["users"] == [] for group in state["groups"]) assert state["mapping_details"][FORGE_GROUP_MAPPING_ID]["name"] == FORGE_OIDC_PROFILE_MAPPING_NAME assert state["mapping_details"][FORGE_GROUP_MAPPING_ID]["expression"] == FORGE_OIDC_PROFILE_MAPPING_EXPRESSION def test_authentik_provider_creation_timeout_is_reconciled_before_application_write( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) remote, writes, _behavior = _install_forge_connector_remote( monkeypatch, provider_timeout_after_commit=True, ) persisted = [] secret = "one-time-client-secret-value-0123456789" result = AuthentikConnector().ensure_forge_oidc_application( client_secret=secret, persist_provider_id=lambda provider_id: persisted.append(provider_id) or True, ) assert result["success"] is True assert result["status"] == "created" assert result["providerId"] == "31" assert [path for _method, path, _payload in writes].count("/providers/oauth2/") == 1 assert [path for _method, path, _payload in writes].count("/core/applications/") == 1 assert persisted == ["31"] assert len(remote["providers"]) == len(remote["applications"]) == 1 assert secret not in str(result) def test_authentik_provider_post_response_must_match_exact_get_readback( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) _remote, writes, _behavior = _install_forge_connector_remote( monkeypatch, provider_response_id="31", provider_created_id="32", ) persisted = [] result = AuthentikConnector().ensure_forge_oidc_application( client_secret="one-time-client-secret-value-0123456789", persist_provider_id=lambda provider_id: persisted.append(provider_id) or True, ) assert result == { "success": False, "status": "partial_failure", "message": "forge_oidc_provider_readback_mismatch", } assert [path for _method, path, _payload in writes] == ["/providers/oauth2/"] assert persisted == [] def test_authentik_application_timeout_retries_only_the_missing_exact_application( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) remote, writes, behavior = _install_forge_connector_remote( monkeypatch, application_fail_without_commit=True, ) def persist(_provider_id): return True secret = "one-time-client-secret-value-0123456789" first = AuthentikConnector().ensure_forge_oidc_application( client_secret=secret, persist_provider_id=persist, ) assert first["status"] == "partial_failure" assert first["message"] == "forge_oidc_application_creation_unknown" assert len(remote["providers"]) == 1 assert remote["applications"] == [] behavior["application_fail_without_commit"] = False retry = AuthentikConnector().ensure_forge_oidc_application( client_secret=secret, persist_provider_id=persist, ) assert retry["success"] is True assert retry["status"] == "created" assert len(remote["providers"]) == 1 assert len(remote["applications"]) == 1 assert [path for _method, path, _payload in writes].count("/providers/oauth2/") == 1 assert [path for _method, path, _payload in writes].count("/core/applications/") == 2 def test_authentik_binding_persistence_failure_keeps_orphan_for_safe_retry( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) remote, writes, _behavior = _install_forge_connector_remote(monkeypatch) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="one-time-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: False, ) assert result == { "success": False, "status": "partial_failure", "message": "forge_oidc_binding_persist_failed", } assert len(remote["providers"]) == 1 assert remote["applications"] == [] assert [path for _method, path, _payload in writes] == ["/providers/oauth2/"] @pytest.mark.parametrize("persistence_result", [1, "true", None]) def test_authentik_provider_binding_persistence_requires_literal_true( monkeypatch: pytest.MonkeyPatch, persistence_result, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) remote, writes, _behavior = _install_forge_connector_remote(monkeypatch) discovery_calls = [] monkeypatch.setattr( AuthentikConnector, "_forge_oidc_discovery_matches", lambda self: discovery_calls.append(True) or True, ) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="one-time-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: persistence_result, ) assert result == { "success": False, "status": "partial_failure", "message": "forge_oidc_binding_persist_failed", } assert len(remote["providers"]) == 1 assert remote["applications"] == [] assert [path for _method, path, _payload in writes] == ["/providers/oauth2/"] assert discovery_calls == [] def test_authentik_get_observation_requires_literal_secret_verification(): from backend.app.connectors.authentik import AuthentikConnector observation = { "success": True, "status": "provider_orphan", "providerId": "31", "secretVerified": True, } assert AuthentikConnector._forge_oidc_result_from_observation( observation, accepted_statuses={"provider_orphan"}, ) is not None for unverified in ( {key: value for key, value in observation.items() if key != "secretVerified"}, {**observation, "secretVerified": False}, {**observation, "secretVerified": 1}, {**observation, "secretVerified": "true"}, ): assert AuthentikConnector._forge_oidc_result_from_observation( unverified, accepted_statuses={"provider_orphan"}, ) is None def test_authentik_application_post_timeout_with_exact_get_pair_is_idempotent( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) remote, writes, _behavior = _install_forge_connector_remote( monkeypatch, application_timeout_after_commit=True, ) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="one-time-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: True, ) assert result["success"] is True assert result["status"] == "already_configured" assert len(remote["providers"]) == len(remote["applications"]) == 1 assert [path for _method, path, _payload in writes].count("/providers/oauth2/") == 1 assert [path for _method, path, _payload in writes].count("/core/applications/") == 1 def test_authentik_provider_detail_secret_comparison_is_constant_time_and_not_retained( monkeypatch: pytest.MonkeyPatch, ): import backend.app.connectors.authentik as authentik_module from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) raw_secret = "remote-only-client-secret-value" requested = [] provider_detail = {"pk": "31", "client_secret": raw_secret} class Response: status_code = 200 def json(self): return provider_detail.copy() class Client: def __init__(self, **kwargs): self.kwargs = kwargs def __enter__(self): return self def __exit__(self, *_args): return False def get(self, url, *, headers): requested.append((url, headers)) return Response() monkeypatch.setattr(authentik_module.httpx, "Client", Client) connector = AuthentikConnector() assert connector._forge_oidc_provider_secret_matches("31", raw_secret) is True assert connector._forge_oidc_provider_secret_matches("31", "different-local-secret") is False provider_detail.clear() provider_detail["pk"] = "31" assert connector._forge_oidc_provider_secret_matches("31", raw_secret) is False assert requested[0][0] == "https://auth.example-host.com/api/v3/providers/oauth2/31/" assert raw_secret not in str(connector.last_error) assert raw_secret not in str(connector._read_cache) def test_authentik_connector_blocks_missing_groups_and_non_rsa_signing_key( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) writes = [] def paged(self, path, params=None): if path == "/core/groups/": return _valid_forge_groups(("forge-viewers", "forge-operators")) if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [ { "pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True, } ] return [] monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_write_request", lambda self, method, path, payload=None: writes.append((method, path, payload)), raising=False, ) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="opaque-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: True, ) assert result["success"] is False assert result["status"] == "blocked" assert result["message"] == "forge_oidc_required_groups_missing" assert writes == [] assert "opaque-client-secret-value-0123456789" not in str(result) def test_authentik_connector_blocks_profile_mapping_that_does_not_emit_filtered_groups_claim( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) invalid_mapping = _forge_group_mapping('return {"roles": ["admin"]}') _mock_group_mapping_api(monkeypatch, invalid_mapping) writes = [] def paged(self, path, params=None): if path == "/core/groups/": return _valid_forge_groups() if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [ { "pk": "rsa-signing-key-id", "key_type": "rsa", "private_key_available": True, } ] if path == "/propertymappings/provider/scope/": mappings = _valid_scope_mappings() mappings[-1] = invalid_mapping return mappings return [] monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_write_request", lambda self, method, path, payload=None: writes.append((method, path, payload)), raising=False, ) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="unused-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: True, ) assert result["status"] == "conflict" assert result["message"] == "forge_oidc_profile_mapping_mismatch" assert writes == [] def test_authentik_connector_refuses_mismatched_existing_resources_without_reconcile( monkeypatch: pytest.MonkeyPatch, ): from backend.app.connectors.authentik import AuthentikConnector _configured_settings(monkeypatch) _mock_group_mapping_api(monkeypatch) writes = [] def paged(self, path, params=None): if path == "/core/groups/": return _valid_forge_groups() if path == "/flows/instances/": return [{"pk": "authorization-flow-id"}, {"pk": "invalidation-flow-id"}] if path == "/crypto/certificatekeypairs/": return [ { "pk": "rsa-signing-key-id", "key_type": {"type": "RSA"}, "private_key_available": True, } ] if path == "/propertymappings/provider/scope/": return _valid_scope_mappings() if path == "/core/applications/": return [ { "pk": "f7212687-a3d0-4eaa-b209-427f4043f870", "slug": "itworx-forge", "meta_launch_url": "https://forge.example.com", "provider": "31", "meta_description": "Managed by AppOps Forge OIDC v1.", } ] if path == "/providers/oauth2/": return [ { "pk": "31", "client_id": "itworx-forge", "client_type": "confidential", "redirect_uris": [ { "matching_mode": "regex", "url": "https://forge.example.com/.*", } ], "authorization_flow": "authorization-flow-id", "invalidation_flow": "invalidation-flow-id", "property_mappings": [FORGE_OPENID_MAPPING_ID, FORGE_EMAIL_MAPPING_ID, FORGE_GROUP_MAPPING_ID], "include_claims_in_id_token": True, "sub_mode": "hashed_user_id", "issuer_mode": "per_provider", "signing_key": "rsa-signing-key-id", } ] raise AssertionError(f"Unexpected Authentik read: {path}") monkeypatch.setattr(AuthentikConnector, "_paged", paged, raising=True) monkeypatch.setattr( AuthentikConnector, "_forge_oidc_write_request", lambda self, method, path, payload=None: writes.append((method, path, payload)), raising=False, ) result = AuthentikConnector().ensure_forge_oidc_application( client_secret="opaque-client-secret-value-0123456789", persist_provider_id=lambda _provider_id: True, ) assert result["success"] is False assert result["status"] == "conflict" assert result["message"] == "forge_oidc_existing_contract_mismatch" assert writes == [] assert "opaque-client-secret-value-0123456789" not in str(result) def test_forge_oidc_provision_lock_serializes_separate_worker_processes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.core.config import settings from backend.app.services import forge_oidc_provisioning from backend.app.services.forge_oidc_provisioning import _forge_oidc_process_lock if os.name == "nt": monkeypatch.setattr( forge_oidc_provisioning, "_windows_private_path", lambda *_args, **_kwargs: None, ) lock_root = tmp_path / "forge-oidc-runtime-lock" if os.name != "nt": monkeypatch.setattr(forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", lock_root) monkeypatch.setattr(settings, "data_dir", str(tmp_path)) ready = tmp_path / "worker-ready" acquired = tmp_path / "worker-acquired" code = """\ from pathlib import Path import os import sys from backend.app.services.forge_oidc_provisioning import _forge_oidc_process_lock from backend.app.services import forge_oidc_provisioning if os.name == "nt": forge_oidc_provisioning._windows_private_path = lambda *_args, **_kwargs: None else: forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT = Path(sys.argv[3]) Path(sys.argv[1]).write_text('ready') with _forge_oidc_process_lock(): Path(sys.argv[2]).write_text('acquired') """ environment = os.environ.copy() environment["APPOPS_DATA_DIR"] = str(tmp_path) process = None try: with _forge_oidc_process_lock(): process = subprocess.Popen( [sys.executable, "-c", code, str(ready), str(acquired), str(lock_root)], cwd=Path(__file__).resolve().parents[2], env=environment, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, ) deadline = time.monotonic() + 10 while not ready.exists() and process.poll() is None and time.monotonic() < deadline: time.sleep(0.02) assert ready.exists(), "worker did not start the lock attempt" time.sleep(0.2) assert process.poll() is None assert not acquired.exists(), "worker entered the critical section before the lock was released" _stdout, stderr = process.communicate(timeout=10) assert process.returncode == 0, stderr assert acquired.read_text() == "acquired" finally: if process is not None and process.poll() is None: process.terminate() process.wait(timeout=5) def test_forge_oidc_provision_fails_closed_when_process_lock_is_unavailable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): from backend.app.services import forge_oidc_provisioning engine = _database(tmp_path, monkeypatch) _configured_settings(monkeypatch) authentik = RecordingAuthentik() class UnavailableLock: def __enter__(self): raise forge_oidc_provisioning._ProvisionLockUnavailable def __exit__(self, *_args): return False monkeypatch.setattr( forge_oidc_provisioning, "_forge_oidc_process_lock", UnavailableLock, ) with Session(engine) as session: result = forge_oidc_provisioning.provision_forge_oidc( session, actor_identity="authentik:operator-1", authentik=authentik, ) assert result["status"] == "failed" assert result["message"] == "forge_oidc_provision_lock_unavailable" assert authentik.preflight_calls == 0 assert authentik.calls == [] def test_forge_oidc_thread_lock_wait_is_bounded(monkeypatch: pytest.MonkeyPatch): from backend.app.services import forge_oidc_provisioning monkeypatch.setattr(forge_oidc_provisioning, "_PROVISION_LOCK_TIMEOUT_SECONDS", 0.01) assert forge_oidc_provisioning._SECRET_LOCK.acquire(blocking=False) try: with ( pytest.raises(forge_oidc_provisioning._ProvisionLockUnavailable), forge_oidc_provisioning._forge_oidc_process_lock(), ): pytest.fail("thread lock should not have been acquired") finally: forge_oidc_provisioning._SECRET_LOCK.release() @pytest.mark.skipif(os.name == "nt", reason="POSIX flock regression") def test_forge_oidc_process_lock_wait_is_bounded( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): import errno import fcntl from backend.app.services import forge_oidc_provisioning monkeypatch.setattr(forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", tmp_path / "forge-oidc-runtime-lock") monkeypatch.setattr(forge_oidc_provisioning, "_PROVISION_LOCK_TIMEOUT_SECONDS", 0.01) attempts = 0 def always_busy(_fd: int, operation: int) -> None: nonlocal attempts attempts += 1 if operation & fcntl.LOCK_UN: return raise BlockingIOError(errno.EAGAIN, "lock is busy") monkeypatch.setattr(fcntl, "flock", always_busy) with ( pytest.raises(forge_oidc_provisioning._ProvisionLockUnavailable), forge_oidc_provisioning._forge_oidc_process_lock(), ): pytest.fail("process lock should have timed out") assert attempts >= 1 @pytest.mark.skipif(os.name == "nt", reason="POSIX directory permission regression") @pytest.mark.parametrize("unsafe_location", ["target", "parent"]) def test_forge_oidc_lock_rejects_writable_data_directory_without_creating_lock( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, unsafe_location: str, ): from backend.app.services import forge_oidc_provisioning unsafe_parent = tmp_path / "shared-writable" unsafe_parent.mkdir(mode=0o700) unsafe_parent.chmod(0o777) if unsafe_location == "target": lock_root = unsafe_parent else: lock_root = unsafe_parent / "private-child" lock_root.mkdir(mode=0o700) monkeypatch.setattr(forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", lock_root) with ( pytest.raises(forge_oidc_provisioning._ProvisionLockUnavailable), forge_oidc_provisioning._forge_oidc_process_lock(), ): pytest.fail("writable data directory must not be trusted") assert not (lock_root / forge_oidc_provisioning._PROVISION_LOCK_FILENAME).exists() @pytest.mark.skipif(os.name == "nt", reason="POSIX lock-file metadata regression") @pytest.mark.parametrize("unsafe_kind", ["hardlink", "permissions"]) def test_forge_oidc_lock_rejects_unsafe_existing_lock_file_without_chmod( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, unsafe_kind: str, ): from backend.app.services import forge_oidc_provisioning lock_root = tmp_path / "forge-oidc-runtime-lock" lock_root.mkdir(mode=0o700) lock_path = lock_root / forge_oidc_provisioning._PROVISION_LOCK_FILENAME lock_path.write_bytes(b"") lock_path.chmod(0o600) if unsafe_kind == "hardlink": os.link(lock_path, tmp_path / "outside-link") else: lock_path.chmod(0o666) original_mode = lock_path.stat().st_mode monkeypatch.setattr(forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", lock_root) with ( pytest.raises(forge_oidc_provisioning._ProvisionLockUnavailable), forge_oidc_provisioning._forge_oidc_process_lock(), ): pytest.fail("unsafe lock file must not be trusted") assert lock_path.stat().st_mode == original_mode @pytest.mark.skipif(os.name != "nt", reason="Windows ACL regression") def test_forge_oidc_windows_lock_rejects_untrusted_writer_acl(tmp_path: Path): from backend.app.services.forge_oidc_provisioning import _windows_private_path unsafe_dir = tmp_path / "forge-oidc-untrusted-acl" unsafe_dir.mkdir() result = subprocess.run( [ "icacls", str(unsafe_dir), "/inheritance:r", "/grant:r", "*S-1-1-0:(OI)(CI)M", ], capture_output=True, text=True, check=False, ) if result.returncode != 0: pytest.skip("Windows ACL tool is unavailable in this runner") with pytest.raises(OSError): _windows_private_path(unsafe_dir, require_current_owner=False)