2138 lines
80 KiB
Python
2138 lines
80 KiB
Python
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, "<forge-profile-mapping>", "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)
|