Files

1066 lines
42 KiB
Python

from __future__ import annotations
import json
from contextlib import contextmanager
from copy import deepcopy
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from uuid import uuid4
import pytest
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine, select
class _RecordingAuthentik:
def __init__(self) -> None:
self.calls: list[tuple[str, str]] = []
def _request(self, path, params=None):
self.calls.append(("GET", path))
raise AssertionError("a governance rejection must not contact Authentik")
def _write_request(self, method, path, payload=None):
self.calls.append((method, path))
raise AssertionError("a governance rejection must not contact Authentik")
def _session():
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
return engine, Session(engine)
@contextmanager
def _unlocked():
yield
def test_exact_policy_rejection_precedes_every_remote_reader_request(monkeypatch):
from backend.app.services import forge_controller_reader as reader
engine, session = _session()
connector = _RecordingAuthentik()
lock_entered = False
@contextmanager
def recording_lock():
nonlocal lock_entered
lock_entered = True
yield
monkeypatch.setattr(reader, "_forge_oidc_process_lock", recording_lock)
monkeypatch.setattr(reader, "_exact_policy_allowed", lambda _session: False)
monkeypatch.setattr(
reader,
"secret_storage_status",
lambda: {"productionReady": True},
)
try:
result = reader.provision_forge_controller_reader(
session,
actor_identity="operator@example.test",
authentik=connector,
)
finally:
session.close()
engine.dispose()
assert lock_entered is True
assert result["status"] == "blocked"
assert result["ready"] is False
assert connector.calls == []
APP_ID = "f7212687-a3d0-4eaa-b209-427f4043f870"
PROVIDER_ID = "31"
MAPPING_IDS = {
"openid": "75a34fd6-cc81-40b8-a8c3-2745af9567ab",
"email": "312810bb-11da-47d8-97d4-f2155348eaa2",
"profile": "7aa8c5f8-5e8e-4a1b-9c11-3a4f0a731cff",
}
GROUP_IDS = {
"forge-viewers": "01e9b457-75be-4c43-a276-5ef49eb691f7",
"forge-operators": "6a238dab-1f85-4fce-b20d-c987a9d2924f",
"forge-owners": "a3430246-574b-4e40-9217-59c12b7babca",
}
KEY_ID = "f2a77680-c53b-4aef-a739-25b8c2477027"
ROLE_ID = "08bd36a8-bfc3-4e7f-91b6-cd34ac2e3e54"
OIDC_SECRET = "fixed-oidc-client-secret-for-reader-tests"
ADMIN_TOKEN = "authentik-admin-token-fixture-never-issued-to-reader"
READER_TOKEN = "authentik-reader-token-fixture-0123456789-abcdef"
def _group(name):
from backend.app.schemas.forge_oidc import FORGE_OIDC_GROUPS
return {
"pk": GROUP_IDS[name],
"num_pk": 14 + FORGE_OIDC_GROUPS.index(name),
"name": name,
"is_superuser": False,
"attributes": {},
"parents": [],
"children": [],
"roles": [],
"inherited_roles_obj": [],
"users": [],
}
def _provider():
from backend.app.schemas.forge_oidc import FORGE_OIDC_CALLBACK_URL
return {
"pk": int(PROVIDER_ID),
"name": "ITWorx Forge OIDC",
"client_id": "itworx-forge",
"client_type": "confidential",
"client_secret": OIDC_SECRET,
"redirect_uris": [{"matching_mode": "strict", "url": FORGE_OIDC_CALLBACK_URL}],
"authorization_flow": "forge-authorization-flow",
"invalidation_flow": "forge-invalidation-flow",
"property_mappings": list(MAPPING_IDS.values()),
"include_claims_in_id_token": True,
"sub_mode": "hashed_user_id",
"issuer_mode": "per_provider",
"signing_key": KEY_ID,
}
def _mappings():
from backend.app.schemas.forge_oidc import (
FORGE_OIDC_EMAIL_MAPPING_EXPRESSION_SHA256,
FORGE_OIDC_EMAIL_MAPPING_MANAGED,
FORGE_OIDC_OPENID_MAPPING_EXPRESSION_SHA256,
FORGE_OIDC_OPENID_MAPPING_MANAGED,
FORGE_OIDC_PROFILE_MAPPING_EXPRESSION,
FORGE_OIDC_PROFILE_MAPPING_NAME,
)
return {
"openid": {
"pk": MAPPING_IDS["openid"],
"managed": FORGE_OIDC_OPENID_MAPPING_MANAGED,
"scope_name": "openid",
"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 {}",
"known_sha256": FORGE_OIDC_OPENID_MAPPING_EXPRESSION_SHA256,
},
"email": {
"pk": MAPPING_IDS["email"],
"managed": FORGE_OIDC_EMAIL_MAPPING_MANAGED,
"scope_name": "email",
"expression": 'return {\n "email": request.user.email,\n "email_verified": False\n}',
"known_sha256": FORGE_OIDC_EMAIL_MAPPING_EXPRESSION_SHA256,
},
"profile": {
"pk": MAPPING_IDS["profile"],
"name": FORGE_OIDC_PROFILE_MAPPING_NAME,
"scope_name": "profile",
"expression": FORGE_OIDC_PROFILE_MAPPING_EXPRESSION,
},
}
class _ReaderRemote:
def __init__(self):
from backend.app.schemas.forge_oidc import (
FORGE_OIDC_APPLICATION_NAME,
FORGE_OIDC_LAUNCH_URL,
FORGE_OIDC_MANAGED_DESCRIPTION,
)
self.calls = []
self.mutations = []
self.roles = []
self.users = []
self.tokens = []
self.object_permissions = []
self.catalog_rows = {}
self.raise_after_permission_post = False
self.raise_after_token_post = False
self.fail_membership_readback_once = False
self.raise_after_membership_post = False
self.application = {
"pk": APP_ID,
"name": FORGE_OIDC_APPLICATION_NAME,
"slug": "itworx-forge",
"meta_launch_url": FORGE_OIDC_LAUNCH_URL,
"meta_description": FORGE_OIDC_MANAGED_DESCRIPTION,
"provider": int(PROVIDER_ID),
}
self.unrelated_application = {
"pk": "0bf8a20a-9438-4dcb-93e1-46e4a6d96090",
"name": "Baseline-visible unrelated app",
"slug": "unrelated-baseline-visible",
"meta_launch_url": "https://unrelated.example.test",
"meta_description": "Unrelated baseline-visible application",
"provider": None,
}
self.provider = _provider()
self.mappings = _mappings()
self.groups = [_group(name) for name in GROUP_IDS]
self.admin_token = ADMIN_TOKEN
self.oidc_secret = OIDC_SECRET
self.reader_token = READER_TOKEN
self.private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
now = datetime.now(timezone.utc).replace(tzinfo=None)
subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "reader-test")])
certificate = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(subject)
.public_key(self.private_key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now - timedelta(minutes=1))
.not_valid_after(now + timedelta(days=10))
.sign(self.private_key, hashes.SHA256())
)
self.certificate_pem = certificate.public_bytes(serialization.Encoding.PEM).decode("ascii")
public_numbers = self.private_key.public_key().public_numbers()
self.jwks_key = {
"kty": "RSA",
"alg": "RS256",
"use": "sig",
"kid": "test-signing-kid",
"n": self._base64url(public_numbers.n),
"e": self._base64url(public_numbers.e),
}
self.discovery = {
"issuer": "https://auth.example-host.com/application/o/itworx-forge/",
"response_types_supported": ["code", "id_token"],
"grant_types_supported": ["authorization_code"],
"scopes_supported": ["openid", "email", "profile"],
"id_token_signing_alg_values_supported": ["RS256"],
"jwks_uri": "https://auth.example-host.com/application/o/itworx-forge/jwks/",
}
@staticmethod
def _base64url(value):
import base64
length = (value.bit_length() + 7) // 8
return base64.urlsafe_b64encode(value.to_bytes(length, "big")).decode("ascii").rstrip("=")
@staticmethod
def _collection(items):
return {"results": deepcopy(items), "pagination": {"next": None}}
def _role_users(self, role_id):
return [
user
for user in self.users
if any(str(item.get("pk") if isinstance(item, dict) else item) == role_id for item in user["roles"])
]
def _groups(self):
groups = deepcopy(self.groups)
for group in groups:
group["roles"] = list(group.get("roles", []))
group["inherited_roles_obj"] = list(group.get("inherited_roles_obj", []))
return groups
def admin(self, _connector, method, path, *, params=None, payload=None):
params = dict(params or {})
self.calls.append((method, path, params, deepcopy(payload)))
if method == "POST":
self.mutations.append((method, path, deepcopy(payload)))
if path == "/rbac/roles/":
role = {"pk": ROLE_ID, "name": payload["name"]}
self.roles.append(role)
return deepcopy(role)
if path.startswith("/rbac/permissions/assigned_by_roles/") and path.endswith("/assign/"):
assert set(payload) == {"permissions", "model", "object_pk"}
assert payload["permissions"] and payload["model"] and payload["object_pk"]
codename = payload["permissions"][0]
app_label, model = payload["model"].split(".", maxsplit=1)
row = {"app_label": app_label, "model": model, "codename": codename, "object_pk": str(payload["object_pk"])}
self.object_permissions.append(row)
if self.raise_after_permission_post:
self.raise_after_permission_post = False
from backend.app.services.forge_controller_reader import (
_ReaderFailure,
)
raise _ReaderFailure("failed", "simulated_transport_timeout")
return {"results": [{"id": 1}]}
if path == "/core/users/":
user = {
"pk": 101,
"username": payload["username"],
"name": payload["name"],
"type": payload["type"],
"path": payload["path"],
"attributes": deepcopy(payload["attributes"]),
"is_active": True,
"is_superuser": False,
"groups": [],
"roles": [],
}
self.users.append(user)
return deepcopy(user)
if path.endswith("/add_user/"):
role_id = path.split("/")[-3]
user_id = str(payload["pk"])
user = next(user for user in self.users if str(user["pk"]) == user_id)
user["roles"] = [role_id]
if self.raise_after_membership_post:
self.raise_after_membership_post = False
self.fail_membership_readback_once = True
from backend.app.services.forge_controller_reader import (
_ReaderFailure,
)
raise _ReaderFailure("failed", "simulated_membership_post_timeout")
return {}
if path == "/core/tokens/":
token = {
"pk": str(uuid4()),
"identifier": payload["identifier"],
"intent": payload["intent"],
"user": payload["user"],
"description": payload["description"],
"expiring": payload["expiring"],
}
self.tokens.append(token)
if self.raise_after_token_post:
self.raise_after_token_post = False
from backend.app.services.forge_controller_reader import (
_ReaderFailure,
)
raise _ReaderFailure("failed", "simulated_transport_timeout")
return deepcopy(token)
raise AssertionError(f"unexpected remote POST path: {path}")
if path == "/core/applications/":
return self._collection([self.application])
if path == "/core/applications/itworx-forge/":
return deepcopy(self.application)
if path == f"/providers/oauth2/{PROVIDER_ID}/":
return deepcopy(self.provider)
for scope, mapping_id in MAPPING_IDS.items():
if path == f"/propertymappings/provider/scope/{mapping_id}/":
return deepcopy(self.mappings[scope])
if path == "/core/groups/":
return self._collection(self._groups())
for name, group_id in GROUP_IDS.items():
if path == f"/core/groups/{group_id}/":
return deepcopy(next(group for group in self._groups() if group["name"] == name))
if path == f"/crypto/certificatekeypairs/{KEY_ID}/":
return {"pk": KEY_ID, "key_type": "rsa", "private_key_available": True, "name": "forge-signing"}
if path == "/rbac/roles/":
if "name" in params:
return self._collection([role for role in self.roles if role["name"] == params["name"]])
if "users" in params:
role_ids = {
str(role["pk"])
for role in self.roles
if any(str(_role_pk) == str(role["pk"]) for user in self._role_users(str(role["pk"])) for _role_pk in user["roles"])
}
return self._collection([role for role in self.roles if str(role["pk"]) in role_ids])
return self._collection(self.roles)
if path == f"/rbac/roles/{ROLE_ID}/":
role = next((role for role in self.roles if role["pk"] == ROLE_ID), None)
return deepcopy(role) if role else {}
if path == "/rbac/permissions/roles/":
return self._collection(self.object_permissions)
if path == "/rbac/permissions/":
if "role" in params:
return self._collection([])
app_label = params["content_type__app_label"]
model = params["content_type__model"]
codename = params["codename"]
row = {"app_label": app_label, "model": model, "codename": codename, "name": codename}
return self._collection([row])
if path == "/core/users/":
if "username" in params:
return self._collection([user for user in self.users if user["username"] == params["username"]])
if "roles_by_pk" in params:
if self.fail_membership_readback_once and self.users:
self.fail_membership_readback_once = False
from backend.app.services.forge_controller_reader import (
_ReaderFailure,
)
raise _ReaderFailure("failed", "simulated_membership_readback_interruption")
return self._collection(self._role_users(str(params["roles_by_pk"])))
return self._collection(self.users)
if path.startswith("/core/users/") and path.endswith("/"):
user_id = path.strip("/").split("/")[-1]
user = next((user for user in self.users if str(user["pk"]) == user_id), None)
return deepcopy(user) if user else {}
if path == "/core/tokens/":
return self._collection([token for token in self.tokens if token["identifier"] == params.get("identifier")])
if path == "/core/tokens/itworx-forge-controller-reader-v1/view_key/":
return {"key": self.reader_token}
raise AssertionError(f"unexpected remote GET path: {path}")
def reader(self, _connector, token, path, *, params=None):
self.calls.append(("READER", path, dict(params or {}), token))
if token != self.reader_token or token == self.admin_token:
raise AssertionError("reader verification used the administrative token")
if path == "/core/users/me/":
user = deepcopy(self.users[0])
user["system_permissions"] = []
user["roles"] = [{"pk": ROLE_ID, "name": "ITWorx Forge controller reader v1"}]
return {"user": user}
if path == "/core/applications/":
apps = (
[self.application]
if params.get("search") == "itworx-forge"
else [self.application, self.unrelated_application]
)
return self._collection(apps)
if path == "/core/applications/itworx-forge/":
return deepcopy(self.application)
if path == f"/providers/oauth2/{PROVIDER_ID}/":
return deepcopy(self.provider)
for scope, mapping_id in MAPPING_IDS.items():
if path == f"/propertymappings/provider/scope/{mapping_id}/":
return deepcopy(self.mappings[scope])
if path == "/core/groups/":
return self._collection(self._groups())
if path == f"/crypto/certificatekeypairs/{KEY_ID}/":
return {"pk": KEY_ID, "key_type": "rsa", "private_key_available": True, "name": "forge-signing"}
if path == f"/crypto/certificatekeypairs/{KEY_ID}/view_certificate/":
return {"data": self.certificate_pem}
raise AssertionError(f"unexpected reader GET path: {path}")
def _setup(monkeypatch, tmp_path, *, remote=None):
from backend.app.core.config import settings
from backend.app.schemas.forge_oidc import (
FORGE_OIDC_BINDING_SECRET_REF,
FORGE_OIDC_SECRET_REF,
)
from backend.app.services import forge_controller_reader as reader
from backend.app.services.secret_store import set_secret
engine = create_engine(
f"sqlite:///{tmp_path / 'forge_reader_test.db'}",
connect_args={"check_same_thread": False},
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
monkeypatch.setattr(settings, "environment", "production")
monkeypatch.setattr(settings, "secret_key", "test-production-secret-key-material-0123456789")
monkeypatch.setattr(settings, "secret_key_source", "env")
for key, value in {
"AUTHENTIK_URL": "https://auth.example-host.com",
"AUTHENTIK_TOKEN": ADMIN_TOKEN,
"AUTHENTIK_VERIFY_TLS": True,
"AUTHENTIK_TIMEOUT_SECONDS": "1",
"AUTHENTIK_OIDC_AUTHORIZATION_FLOW": "forge-authorization-flow",
"AUTHENTIK_OIDC_INVALIDATION_FLOW": "forge-invalidation-flow",
"AUTHENTIK_FORGE_OIDC_SIGNING_KEY_ID": KEY_ID,
"AUTHENTIK_FORGE_OIDC_PROVIDER_ID": PROVIDER_ID,
"AUTHENTIK_FORGE_OIDC_BINDING_VALID": True,
"AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING": MAPPING_IDS["openid"],
"AUTHENTIK_FORGE_OIDC_EMAIL_CLAIM_MAPPING": MAPPING_IDS["email"],
"AUTHENTIK_FORGE_OIDC_PROFILE_CLAIM_MAPPING": MAPPING_IDS["profile"],
}.items():
monkeypatch.setitem(settings.apis, key, value)
set_secret(
session,
FORGE_OIDC_BINDING_SECRET_REF,
json.dumps({
"version": 1,
"providerId": PROVIDER_ID,
"scopeMappingIds": MAPPING_IDS,
}, separators=(",", ":"), sort_keys=True),
)
set_secret(session, FORGE_OIDC_SECRET_REF, OIDC_SECRET)
remote = remote or _ReaderRemote()
connector = SimpleNamespace(
_config=lambda: SimpleNamespace(
url="https://auth.example-host.com",
verify_tls=True,
token=ADMIN_TOKEN,
timeout_seconds=1,
)
)
monkeypatch.setattr(reader, "_forge_oidc_process_lock", _unlocked)
monkeypatch.setattr(reader, "_exact_policy_allowed", lambda _session: True)
monkeypatch.setattr(reader, "secret_storage_status", lambda: {"productionReady": True})
monkeypatch.setattr(reader, "_admin_json", remote.admin)
monkeypatch.setattr(reader, "_reader_json", remote.reader)
monkeypatch.setattr(
reader,
"_public_json",
lambda url: deepcopy(remote.discovery) if url.endswith(".well-known/openid-configuration") else {"keys": [deepcopy(remote.jwks_key)]},
)
return engine, session, connector, remote, reader
def _provision(session, connector, reader):
return reader.provision_forge_controller_reader(
session,
actor_identity="operator@example.test",
authentik=connector,
)
def test_default_reader_connector_loads_only_after_locked_governance_and_store_checks(
monkeypatch,
tmp_path,
):
from backend.app.services import forge_controller_reader as reader
from backend.app.services import setup_service
engine, session, _connector, remote, _reader = _setup(monkeypatch, tmp_path)
events = []
loader_calls = []
@contextmanager
def recording_lock():
events.append("lock-enter")
yield
events.append("lock-exit")
def policy_check(_session):
events.append("policy")
return True
def store_status():
events.append("store")
return {"productionReady": True}
original_loader = setup_service.apply_stored_connector_settings
def tracked_loader(current_session, connector_ids=None):
events.append("loader")
loader_calls.append(connector_ids)
return original_loader(current_session, connector_ids)
class FakeAuthentikConnector:
pass
connector_seen = []
def provision_locked(_session, connector):
events.append("provision")
connector_seen.append(connector)
return reader._result("ready", "forge_reader_ready", ready=True)
monkeypatch.setattr(reader, "_exact_policy_allowed", policy_check)
monkeypatch.setattr(reader, "secret_storage_status", store_status)
monkeypatch.setattr(reader, "_forge_oidc_process_lock", recording_lock)
monkeypatch.setattr(setup_service, "apply_stored_connector_settings", tracked_loader)
monkeypatch.setattr(reader, "AuthentikConnector", FakeAuthentikConnector)
monkeypatch.setattr(reader, "_provision_locked", provision_locked)
try:
result = reader.provision_forge_controller_reader(
session,
actor_identity="operator@example.test",
)
assert result["status"] == "ready"
assert events == [
"lock-enter",
"policy",
"store",
"loader",
"provision",
"lock-exit",
]
assert loader_calls == [{"authentik"}]
assert len(connector_seen) == 1
assert isinstance(connector_seen[0], FakeAuthentikConnector)
assert remote.calls == []
assert remote.mutations == []
finally:
session.close()
engine.dispose()
def test_default_reader_connector_is_not_loaded_when_store_is_not_production_ready(
monkeypatch,
tmp_path,
):
from backend.app.services import forge_controller_reader as reader
from backend.app.services import setup_service
engine, session, _connector, remote, _reader = _setup(monkeypatch, tmp_path)
loader_calls = []
original_loader = setup_service.apply_stored_connector_settings
def tracked_loader(current_session, connector_ids=None):
loader_calls.append(connector_ids)
return original_loader(current_session, connector_ids)
monkeypatch.setattr(setup_service, "apply_stored_connector_settings", tracked_loader)
monkeypatch.setattr(reader, "_exact_policy_allowed", lambda _session: True)
monkeypatch.setattr(reader, "secret_storage_status", lambda: {"productionReady": False})
try:
result = reader.provision_forge_controller_reader(
session,
actor_identity="operator@example.test",
)
assert result["status"] == "blocked"
assert result["message"] == "forge_reader_secret_store_not_production_ready"
assert loader_calls == []
assert remote.calls == []
assert remote.mutations == []
finally:
session.close()
engine.dispose()
def test_fixed_reader_creates_exact_object_permission_set_and_reconciles_ready_retry(monkeypatch, tmp_path):
from backend.app.db.models import AuditEvent, Secret
from backend.app.services.secret_store import get_secret
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
try:
result = _provision(session, connector, reader)
assert result == {
"success": True,
"status": "ready",
"message": "forge_reader_ready",
"ready": True,
"secretRef": "ITWORX_FORGE_AUTHENTIK_READ_TOKEN",
}
assert len(remote.mutations) == 14
assignments = [item for item in remote.mutations if "/assign/" in item[1]]
assert len(assignments) == 10
assert all(
set(payload) == {"permissions", "model", "object_pk"}
and len(payload["permissions"]) == 1
and payload["model"]
and payload["object_pk"]
for _, _, payload in assignments
)
assert len(remote.object_permissions) == 10
assert remote.tokens[0]["intent"] == "api"
assert remote.tokens[0]["expiring"] is False
assert remote.users[0]["username"] == "itworx-forge-controller-reader"
assert remote.users[0]["type"] == "service_account"
assert remote.users[0]["groups"] == []
assert "password" not in next(payload for method, path, payload in remote.mutations if method == "POST" and path == "/core/users/")
reader_calls = [call for call in remote.calls if call[0] == "READER"]
assert reader_calls and {call[3] for call in reader_calls} == {READER_TOKEN}
assert all(call[3] != ADMIN_TOKEN for call in reader_calls)
application_lists = [call for call in reader_calls if call[1] == "/core/applications/"]
assert application_lists and all(call[2].get("search") == "itworx-forge" for call in application_lists)
assert not any(
call[1] == "/core/applications/unrelated-baseline-visible/"
for call in reader_calls
)
assert ("GET", "/core/applications/itworx-forge/") in {
(call[0], call[1]) for call in remote.calls if call[0] == "GET"
}
assert ("READER", "/core/applications/itworx-forge/") in {
(call[0], call[1]) for call in reader_calls
}
token = get_secret(session, reader.FORGE_READER_TOKEN_REF, touch=False)
assert token == READER_TOKEN
binding = json.loads(get_secret(session, reader.FORGE_READER_BINDING_REF, touch=False))
first_nonce = binding["nonce"]
assert binding["phase"] == "ready"
assert binding["userId"] == "101"
assert binding["roleId"] == ROLE_ID
assert binding["tokenId"] == remote.tokens[0]["pk"]
assert all(token not in row.encrypted_value for row in session.exec(select(Secret)).all())
assert OIDC_SECRET not in json.dumps(result)
assert ADMIN_TOKEN not in json.dumps(result)
assert READER_TOKEN not in json.dumps(result)
audit = session.exec(select(AuditEvent)).all()
assert audit and all(READER_TOKEN not in json.dumps(row.redacted_metadata or {}) for row in audit)
mutation_count = len(remote.mutations)
token_count = len(remote.tokens)
retry = _provision(session, connector, reader)
next_binding = json.loads(get_secret(session, reader.FORGE_READER_BINDING_REF, touch=False))
assert retry["ready"] is True, retry
assert len(remote.mutations) == mutation_count
assert len(remote.tokens) == token_count == 1
assert next_binding["nonce"] == first_nonce
assert get_secret(session, reader.FORGE_READER_TOKEN_REF, touch=False) == READER_TOKEN
finally:
session.close()
engine.dispose()
def test_unknown_permission_post_is_reconciled_by_get_without_replaying_it(monkeypatch, tmp_path):
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
remote.raise_after_permission_post = True
try:
result = _provision(session, connector, reader)
assert result["ready"] is True
assignments = [item for item in remote.mutations if "/assign/" in item[1]]
assert len(assignments) == 10
assert len(remote.object_permissions) == 10
finally:
session.close()
engine.dispose()
def test_journal_persistence_failure_prevents_first_remote_mutation(monkeypatch, tmp_path):
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
monkeypatch.setattr(reader, "_persist_journal", lambda *_args: False)
try:
result = _provision(session, connector, reader)
assert result["status"] == "partial_failure"
assert result["ready"] is False
assert not remote.mutations
finally:
session.close()
engine.dispose()
def test_existing_unowned_user_is_left_untouched_before_any_write(monkeypatch, tmp_path):
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
remote.users.append({
"pk": 555,
"username": "itworx-forge-controller-reader",
"name": "unrelated account",
"type": "service_account",
"path": "service-accounts/itworx-forge",
"attributes": {},
"is_active": True,
"is_superuser": False,
"groups": [],
"roles": [],
})
try:
result = _provision(session, connector, reader)
assert result["status"] == "conflict"
assert result["ready"] is False
assert not remote.mutations
assert remote.users[0]["name"] == "unrelated account"
finally:
session.close()
engine.dispose()
@pytest.mark.parametrize("bad_version", [True, 1.0, 2])
def test_nonliteral_or_unsupported_journal_version_conflicts_before_mutations(monkeypatch, tmp_path, bad_version):
from backend.app.services.secret_store import set_secret
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
journal = {
"version": bad_version,
"nonce": str(uuid4()),
"resourceContext": {},
"permissionManifest": [],
"userId": None,
"roleId": None,
"tokenId": None,
"phase": "role",
"intent": "role",
"permissionIndex": None,
}
set_secret(session, reader.FORGE_READER_BINDING_REF, json.dumps(journal, separators=(",", ":"), sort_keys=True))
try:
result = _provision(session, connector, reader)
assert result["status"] == "conflict"
assert not remote.mutations
finally:
session.close()
engine.dispose()
def test_original_token_is_recovered_after_token_store_failure_without_rotation(monkeypatch, tmp_path):
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
persist_token = reader._persist_token
monkeypatch.setattr(reader, "_persist_token", lambda *_args: False)
try:
result = _provision(session, connector, reader)
assert result["status"] == "partial_failure"
assert result["ready"] is False
assert len(remote.tokens) == 1
token_posts = sum(path == "/core/tokens/" for method, path, _payload in remote.mutations if method == "POST")
assert token_posts == 1
monkeypatch.setattr(reader, "_persist_token", persist_token)
retry = _provision(session, connector, reader)
token_posts_after_retry = sum(path == "/core/tokens/" for method, path, _payload in remote.mutations if method == "POST")
assert retry["ready"] is True, retry
assert len(remote.tokens) == 1
assert token_posts_after_retry == 1
finally:
session.close()
engine.dispose()
def test_membership_intent_recovers_when_interrupted_before_assignment_post(monkeypatch, tmp_path):
from backend.app.services.secret_store import get_secret
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
remote.fail_membership_readback_once = True
try:
interrupted = _provision(session, connector, reader)
binding = json.loads(get_secret(session, reader.FORGE_READER_BINDING_REF, touch=False))
membership_posts = [
row for row in remote.mutations if row[1].endswith("/add_user/")
]
assert interrupted["ready"] is False
assert binding["phase"] == "membership"
assert binding["intent"] == "membership"
assert binding["userId"] == "101"
assert remote.users[0]["roles"] == []
assert membership_posts == []
recovered = _provision(session, connector, reader)
membership_posts = [
row for row in remote.mutations if row[1].endswith("/add_user/")
]
assert recovered["ready"] is True, recovered
assert len(remote.users) == 1
assert remote.users[0]["roles"] == [ROLE_ID]
assert len(remote.roles) == 1
assert len(membership_posts) == 1
assert all(not group["roles"] for group in remote.groups)
finally:
session.close()
engine.dispose()
def test_membership_intent_recovers_when_post_applies_before_transport_loss(monkeypatch, tmp_path):
from backend.app.services.secret_store import get_secret
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
remote.raise_after_membership_post = True
try:
interrupted = _provision(session, connector, reader)
binding = json.loads(get_secret(session, reader.FORGE_READER_BINDING_REF, touch=False))
membership_posts = [
row for row in remote.mutations if row[1].endswith("/add_user/")
]
assert interrupted["ready"] is False
assert binding["phase"] == "membership"
assert binding["intent"] == "membership"
assert remote.users[0]["roles"] == [ROLE_ID]
assert len(membership_posts) == 1
assert remote.tokens == []
recovered = _provision(session, connector, reader)
membership_posts = [
row for row in remote.mutations if row[1].endswith("/add_user/")
]
assert recovered["ready"] is True, recovered
assert len(remote.users) == 1
assert remote.users[0]["roles"] == [ROLE_ID]
assert len(remote.roles) == 1
assert len(membership_posts) == 1
assert len(remote.tokens) == 1
assert all(not group["roles"] for group in remote.groups)
finally:
session.close()
engine.dispose()
def test_collection_with_malformed_rows_fails_closed():
from backend.app.services import forge_controller_reader as reader
with pytest.raises(reader._ReaderFailure) as failure:
reader._json_items({"results": [{"pk": "valid"}, "not-an-object"], "pagination": {"next": None}})
assert failure.value.code == "forge_reader_remote_collection_invalid"
def test_secret_delivery_verifier_requires_ready_reader_and_performs_gets_only(
monkeypatch, tmp_path
):
from backend.app.services.secret_store import get_secret
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
try:
assert _provision(session, connector, reader)["ready"] is True
before_mutations = deepcopy(remote.mutations)
remote.calls.clear()
verified = reader._verify_forge_delivery_inputs(session, connector)
assert verified.oidc_client_secret == OIDC_SECRET.encode("ascii")
assert verified.authentik_read_token == READER_TOKEN.encode("ascii")
assert verified.context == {
"applicationId": APP_ID,
"providerId": PROVIDER_ID,
"scopeMappingIds": MAPPING_IDS,
"groupIds": GROUP_IDS,
"signingKeyId": KEY_ID,
}
assert OIDC_SECRET not in repr(verified)
assert READER_TOKEN not in repr(verified)
assert remote.mutations == before_mutations
assert remote.calls
assert all(call[0] in {"GET", "READER"} for call in remote.calls)
assert get_secret(session, reader.FORGE_READER_TOKEN_REF, touch=False) == READER_TOKEN
finally:
session.close()
engine.dispose()
def test_secret_delivery_verifier_rejects_partial_journal_before_remote_reads(
monkeypatch, tmp_path
):
from backend.app.services.secret_store import get_secret, set_secret
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
try:
assert _provision(session, connector, reader)["ready"] is True
journal = json.loads(get_secret(session, reader.FORGE_READER_BINDING_REF, touch=False))
journal["phase"] = "token"
journal["intent"] = "token"
set_secret(
session,
reader.FORGE_READER_BINDING_REF,
json.dumps(journal, separators=(",", ":"), sort_keys=True),
)
before_mutations = deepcopy(remote.mutations)
remote.calls.clear()
with pytest.raises(reader._ReaderFailure) as failure:
reader._verify_forge_delivery_inputs(session, connector)
assert failure.value.code == "forge_reader_binding_not_ready"
assert remote.calls == []
assert remote.mutations == before_mutations
finally:
session.close()
engine.dispose()
@pytest.mark.parametrize("drift", ["extra_permission", "token_expiring", "user_privileged"])
def test_secret_delivery_verifier_rejects_reader_privilege_or_token_drift_without_writes(
monkeypatch, tmp_path, drift
):
engine, session, connector, remote, reader = _setup(monkeypatch, tmp_path)
try:
assert _provision(session, connector, reader)["ready"] is True
if drift == "extra_permission":
remote.object_permissions.append(
{
"app_label": "authentik_core",
"model": "user",
"codename": "view_user",
"object_pk": "101",
}
)
elif drift == "token_expiring":
remote.tokens[0]["expiring"] = True
else:
remote.users[0]["is_superuser"] = True
before_mutations = deepcopy(remote.mutations)
remote.calls.clear()
with pytest.raises(reader._ReaderFailure):
reader._verify_forge_delivery_inputs(session, connector)
assert remote.mutations == before_mutations
assert remote.calls
assert all(call[0] in {"GET", "READER"} for call in remote.calls)
finally:
session.close()
engine.dispose()
@pytest.mark.parametrize(
"next_value",
[
"https://auth.example-host.com/api/v3/core/users/?page=2&page_size=100",
"https://auth.example-host.com/api/v3/core/applications/?page=2&page_size=99",
"https://user@auth.example-host.com/api/v3/core/applications/?page=2&page_size=100",
],
)
def test_pagination_rejects_endpoint_filter_or_authority_drift(next_value):
from backend.app.services import forge_controller_reader as reader
requests = []
def request(path, *, params=None):
requests.append((path, params))
return {"results": [{"pk": APP_ID}], "pagination": {"next": next_value}}
with pytest.raises(reader._ReaderFailure) as failure:
reader._paged(request, "/core/applications/", {"page_size": 100})
assert failure.value.code == "forge_reader_remote_pagination_invalid"
assert len(requests) == 1
@pytest.mark.parametrize("second_next", ["1", "2"])
def test_pagination_rejects_backward_or_repeated_pages(second_next):
from backend.app.services import forge_controller_reader as reader
requests = []
def request(path, *, params=None):
requests.append((path, params))
next_value = "2" if len(requests) == 1 else second_next
return {"results": [], "pagination": {"next": next_value}}
with pytest.raises(reader._ReaderFailure) as failure:
reader._paged(request, "/core/applications/", {"page_size": 100})
assert failure.value.code == "forge_reader_remote_pagination_invalid"
assert requests == [
("/core/applications/", {"page_size": 100}),
("/core/applications/", {"page_size": 100, "page": 2}),
]
def test_http_json_clamps_timeout_and_rejects_oversized_stream_without_echoing_body(monkeypatch):
from backend.app.services import forge_controller_reader as reader
secret_text = "never-echo-this-reader-token"
observed = {}
class _Response:
status_code = 200
headers = {}
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def iter_bytes(self, *, chunk_size):
assert chunk_size == 16 * 1024
body = b"x" * (reader._MAX_RESPONSE_BYTES + 1) + secret_text.encode()
for index in range(0, len(body), chunk_size):
yield body[index : index + chunk_size]
class _Client:
def __init__(self, *, timeout, verify, follow_redirects):
observed["timeout"] = timeout
assert verify is True
assert follow_redirects is False
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def stream(self, method, url, **kwargs):
assert method == "GET"
assert url == "https://auth.example-host.com/test"
assert kwargs["headers"] == {"Accept": "application/json"}
return _Response()
monkeypatch.setattr(reader.httpx, "Client", _Client)
with pytest.raises(reader._ReaderFailure) as failure:
reader._http_json(
"GET",
"https://auth.example-host.com/test",
token=None,
timeout=9999,
)
assert observed["timeout"] == reader._HTTP_TIMEOUT_MAX_SECONDS
assert failure.value.code == "forge_reader_remote_response_too_large"
assert secret_text not in str(failure.value)