from __future__ import annotations import base64 import hmac import json import math import re from dataclasses import dataclass, field from typing import Any from urllib.parse import parse_qs, quote, urlparse from uuid import RFC_4122, UUID, uuid4 import httpx from cryptography import x509 from cryptography.hazmat.primitives.asymmetric import rsa from sqlmodel import Session from ..connectors.authentik import ( AuthentikConnector, _forge_oidc_group_contract_matches, _forge_oidc_scope_mapping_matches, ) from ..core.config import settings from ..schemas.forge_oidc import ( FORGE_OIDC_ACTION, FORGE_OIDC_APPLICATION_NAME, FORGE_OIDC_BINDING_SECRET_REF, FORGE_OIDC_CLIENT_ID, FORGE_OIDC_DISCOVERY_URL, FORGE_OIDC_GROUPS, FORGE_OIDC_ISSUER, FORGE_OIDC_LAUNCH_URL, FORGE_OIDC_MANAGED_DESCRIPTION, FORGE_OIDC_SECRET_REF, FORGE_OIDC_TARGET, ForgeOidcBindingV1, ForgeOidcScopeMappingIdsV1, ) from .audit_logger import log_event from .forge_oidc_provisioning import _exact_policy_allowed, _forge_oidc_process_lock from .secret_store import get_secret, has_secret, secret_storage_status, set_secret from .setup_service import _apply_forge_oidc_binding_to_runtime FORGE_READER_BINDING_REF = "AUTHENTIK_FORGE_READER_BINDING" FORGE_READER_TOKEN_REF = "ITWORX_FORGE_AUTHENTIK_READ_TOKEN" FORGE_READER_USERNAME = "itworx-forge-controller-reader" FORGE_READER_NAME = "ITWorx Forge controller reader v1" FORGE_READER_PATH = "service-accounts/itworx-forge" FORGE_READER_ROLE = "ITWorx Forge controller reader v1" FORGE_READER_TOKEN_ID = "itworx-forge-controller-reader-v1" FORGE_READER_TOKEN_DESCRIPTION = "Managed by AppOps Forge controller reader v1." FORGE_READER_OWNER_ATTRIBUTE = "appops_forge_controller_reader_v1" _AUTHENTIK_BASE = "https://auth.example-host.com" _JOURNAL_KEYS = { "version", "nonce", "resourceContext", "permissionManifest", "userId", "roleId", "tokenId", "phase", "intent", "permissionIndex", } _RESOURCE_KEYS = {"applicationId", "providerId", "scopeMappingIds", "groupIds", "signingKeyId"} _PHASES = {"role", "permissions", "user", "membership", "token", "ready"} _INTENTS = {None, "role", "permission", "user", "membership", "token"} _TOKEN_PATTERN = re.compile(r"[\x21-\x7e]{32,256}\Z") _USER_ID_PATTERN = re.compile(r"[1-9][0-9]{0,18}\Z") _PROVIDER_ID_PATTERN = re.compile(r"[1-9][0-9]{0,9}\Z") _GROUP_LIST_PARAMS = { "include_users": "false", "include_children": "true", "include_parents": "true", "include_inherited_roles": "true", "page_size": 100, } _MAX_RESPONSE_BYTES = 256 * 1024 _HTTP_TIMEOUT_MIN_SECONDS = 0.1 _HTTP_TIMEOUT_MAX_SECONDS = 10.0 class _ReaderFailure(RuntimeError): def __init__(self, status: str, code: str): super().__init__(code) self.status = status self.code = code def _fail(code: str, *, status: str = "conflict") -> None: raise _ReaderFailure(status, code) def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ValueError("duplicate_json_key") result[key] = value return result def _uuid(value: Any, *, version4: bool = False) -> str | None: if not isinstance(value, str): return None try: parsed = UUID(value) except (TypeError, ValueError, AttributeError): return None if ( str(parsed) != value or parsed.variant != RFC_4122 or parsed.version not in range(1, 9) or (version4 and parsed.version != 4) ): return None return value def _positive_id(value: Any, *, provider: bool = False) -> str | None: if type(value) is int: candidate = str(value) elif isinstance(value, str): candidate = value else: return None pattern = _PROVIDER_ID_PATTERN if provider else _USER_ID_PATTERN return candidate if pattern.fullmatch(candidate) else None def _remote_id(value: Any) -> Any: if isinstance(value, dict): return value.get("pk", value.get("id")) return value def _json_items(payload: Any) -> tuple[list[dict[str, Any]], str | None]: if isinstance(payload, list): if any(not isinstance(item, dict) for item in payload): _fail("forge_reader_remote_collection_invalid", status="failed") return (payload, None) if not isinstance(payload, dict): _fail("forge_reader_remote_collection_invalid", status="failed") collection_keys = [key for key in ("results", "data", "items") if key in payload] if len(collection_keys) != 1: _fail("forge_reader_remote_collection_invalid", status="failed") items = payload[collection_keys[0]] if not isinstance(items, list) or any(not isinstance(item, dict) for item in items): _fail("forge_reader_remote_collection_invalid", status="failed") pagination = payload.get("pagination") if "pagination" in payload and not isinstance(pagination, dict): _fail("forge_reader_remote_collection_invalid", status="failed") nested_next = pagination.get("next") if isinstance(pagination, dict) and "next" in pagination else None direct_next = payload.get("next") if "next" in payload else None if "next" in payload and isinstance(pagination, dict) and "next" in pagination and direct_next != nested_next: _fail("forge_reader_remote_collection_invalid", status="failed") next_value = direct_next if "next" in payload else nested_next if next_value is None: return items, None if type(next_value) is int and next_value > 0: next_value = str(next_value) if not isinstance(next_value, str) or not next_value: _fail("forge_reader_remote_collection_invalid", status="failed") return items, next_value def _bounded_timeout(value: Any) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): _fail("forge_reader_timeout_invalid", status="blocked") try: seconds = float(value) except (OverflowError, ValueError): _fail("forge_reader_timeout_invalid", status="blocked") if not math.isfinite(seconds): _fail("forge_reader_timeout_invalid", status="blocked") return min(max(seconds, _HTTP_TIMEOUT_MIN_SECONDS), _HTTP_TIMEOUT_MAX_SECONDS) def _http_json( method: str, url: str, *, token: str | None, timeout: float, params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, ) -> Any: headers = {"Accept": "application/json"} if token is not None: headers["Authorization"] = f"Bearer {token}" try: with httpx.Client(timeout=_bounded_timeout(timeout), verify=True, follow_redirects=False) as client: with client.stream(method, url, params=params, json=payload, headers=headers) as response: if response.status_code < 200 or response.status_code >= 300: _fail("forge_reader_remote_request_failed", status="failed") content_length = response.headers.get("content-length") if content_length is not None: if not content_length.isascii() or not content_length.isdecimal() or int(content_length) > _MAX_RESPONSE_BYTES: _fail("forge_reader_remote_response_too_large", status="failed") chunks: list[bytes] = [] total_bytes = 0 for chunk in response.iter_bytes(chunk_size=16 * 1024): total_bytes += len(chunk) if total_bytes > _MAX_RESPONSE_BYTES: _fail("forge_reader_remote_response_too_large", status="failed") chunks.append(chunk) response_body = b"".join(chunks) chunks.clear() if not response_body: response_body = None return {} try: value = json.loads(response_body) except (ValueError, TypeError, UnicodeError): response_body = None _fail("forge_reader_remote_response_invalid", status="failed") response_body = None return value except _ReaderFailure: raise except (httpx.RequestError, OSError): _fail("forge_reader_remote_unavailable", status="failed") def _admin_json( connector: Any, method: str, path: str, *, params: dict[str, Any] | None = None, payload: dict[str, Any] | None = None, ) -> Any: config = connector._config() if ( config is None or str(getattr(config, "url", "")).rstrip("/") != _AUTHENTIK_BASE or getattr(config, "verify_tls", None) is not True or not isinstance(getattr(config, "token", None), str) or not config.token or not path.startswith("/") or path.startswith("//") ): _fail("forge_reader_authentik_configuration_required", status="blocked") return _http_json( method, f"{_AUTHENTIK_BASE}/api/v3{path}", token=config.token, timeout=config.timeout_seconds, params=params, payload=payload, ) def _reader_json(connector: Any, reader_token: str, path: str, *, params=None) -> Any: config = connector._config() if ( config is None or str(getattr(config, "url", "")).rstrip("/") != _AUTHENTIK_BASE or getattr(config, "verify_tls", None) is not True or not path.startswith("/") or path.startswith("//") ): _fail("forge_reader_authentik_configuration_required", status="blocked") return _http_json( "GET", f"{_AUTHENTIK_BASE}/api/v3{path}", token=reader_token, timeout=config.timeout_seconds, params=params, ) def _public_json(url: str) -> Any: parsed = urlparse(url) if parsed.scheme != "https" or parsed.netloc != urlparse(_AUTHENTIK_BASE).netloc: _fail("forge_reader_public_metadata_url_invalid", status="conflict") return _http_json("GET", url, token=None, timeout=10.0) def _paged( request, path: str, params: dict[str, Any] | None = None, *, max_pages: int = 100, max_items: int = 10000, ) -> list[dict[str, Any]]: base_params = dict(params or {}) if not isinstance(path, str) or not path.startswith("/") or path.startswith("//") or urlparse(path).path != path: _fail("forge_reader_remote_pagination_invalid", status="failed") if "page" in base_params: _fail("forge_reader_remote_pagination_invalid", status="failed") base_query = {key: [str(value)] for key, value in base_params.items()} page_path = path page_params: dict[str, Any] | None = dict(base_params) items: list[dict[str, Any]] = [] current_page = 1 seen_pages = {current_page} for _ in range(max_pages): response = request(page_path, params=page_params) page_items, next_value = _json_items(response) items.extend(page_items) if len(items) > max_items: _fail("forge_reader_remote_collection_limit", status="failed") if not next_value: return items if re.fullmatch(r"[1-9][0-9]*", next_value): next_page = int(next_value) else: if "#" in next_value: _fail("forge_reader_remote_pagination_invalid", status="failed") parsed = urlparse(next_value) if parsed.username is not None or parsed.password is not None or parsed.fragment: _fail("forge_reader_remote_pagination_invalid", status="failed") if parsed.scheme or parsed.netloc: expected_host = urlparse(_AUTHENTIK_BASE) if parsed.scheme != "https" or parsed.netloc != expected_host.netloc: _fail("forge_reader_remote_pagination_invalid", status="failed") elif next_value.startswith("//"): _fail("forge_reader_remote_pagination_invalid", status="failed") expected_paths = {path, f"/api/v3{path}"} if parsed.path not in expected_paths: _fail("forge_reader_remote_pagination_invalid", status="failed") try: query_params = parse_qs(parsed.query, keep_blank_values=True, strict_parsing=True) except ValueError: _fail("forge_reader_remote_pagination_invalid", status="failed") if set(query_params) != set(base_query) | {"page"} or any( query_params.get(key) != values for key, values in base_query.items() ): _fail("forge_reader_remote_pagination_invalid", status="failed") page_values = query_params.get("page", []) if len(page_values) != 1 or not re.fullmatch(r"[1-9][0-9]*", page_values[0]): _fail("forge_reader_remote_pagination_invalid", status="failed") next_page = int(page_values[0]) if next_page < 1 or next_page > max_pages or next_page <= current_page or next_page in seen_pages: _fail("forge_reader_remote_pagination_invalid", status="failed") current_page = next_page seen_pages.add(current_page) page_path = path page_params = {**base_params, "page": current_page} _fail("forge_reader_remote_pagination_incomplete", status="failed") def _manifest(context: dict[str, Any]) -> list[dict[str, str]]: app_id = context["applicationId"] provider_id = context["providerId"] mappings = context["scopeMappingIds"] groups = context["groupIds"] key_id = context["signingKeyId"] rows = [ {"app": "authentik_core", "model": "application", "codename": "view_application", "objectPk": app_id}, {"app": "authentik_providers_oauth2", "model": "oauth2provider", "codename": "view_oauth2provider", "objectPk": provider_id}, *[ { "app": "authentik_providers_oauth2", "model": "scopemapping", "codename": "view_scopemapping", "objectPk": mappings[scope], } for scope in ("openid", "email", "profile") ], *[ { "app": "authentik_core", "model": "group", "codename": "view_group", "objectPk": groups[name], } for name in FORGE_OIDC_GROUPS ], {"app": "authentik_crypto", "model": "certificatekeypair", "codename": "view_certificatekeypair", "objectPk": key_id}, { "app": "authentik_crypto", "model": "certificatekeypair", "codename": "view_certificatekeypair_certificate", "objectPk": key_id, }, ] return sorted(rows, key=lambda row: (row["app"], row["model"], row["codename"], row["objectPk"])) def _journal_bytes(journal: dict[str, Any]) -> str: return json.dumps(journal, ensure_ascii=True, separators=(",", ":"), sort_keys=True) def _valid_journal(value: Any) -> bool: if not isinstance(value, dict) or set(value) != _JOURNAL_KEYS: return False if type(value.get("version")) is not int or value["version"] != 1: return False if _uuid(value.get("nonce"), version4=True) is None: return False context = value.get("resourceContext") if not isinstance(context, dict) or set(context) != _RESOURCE_KEYS: return False if ( _uuid(context.get("applicationId")) is None or not isinstance(context.get("providerId"), str) or _positive_id(context.get("providerId"), provider=True) is None or _uuid(context.get("signingKeyId")) is None ): return False try: mapping_ids = ForgeOidcScopeMappingIdsV1.model_validate(context.get("scopeMappingIds")) except Exception: # noqa: BLE001 - journal is untrusted encrypted state. return False if not isinstance(context.get("groupIds"), dict) or set(context["groupIds"]) != set(FORGE_OIDC_GROUPS): return False if any(_uuid(context["groupIds"][name]) is None for name in FORGE_OIDC_GROUPS): return False if len({str(mapping_ids.openid), str(mapping_ids.email), str(mapping_ids.profile)}) != 3: return False if len(set(context["groupIds"].values())) != 3: return False if not isinstance(value.get("permissionManifest"), list) or len(value["permissionManifest"]) != 10: return False for row in value["permissionManifest"]: if ( not isinstance(row, dict) or set(row) != {"app", "model", "codename", "objectPk"} or any(not isinstance(row[key], str) or not row[key] for key in row) ): return False user_id = value.get("userId") if user_id is not None and (not isinstance(user_id, str) or _positive_id(user_id) is None): return False for key in ("roleId", "tokenId"): if value.get(key) is not None and _uuid(value[key]) is None: return False if not isinstance(value.get("phase"), str) or value["phase"] not in _PHASES: return False if value.get("intent") is not None and ( not isinstance(value.get("intent"), str) or value["intent"] not in _INTENTS ): return False index = value.get("permissionIndex") if index is not None and (type(index) is not int or not 0 <= index < 10): return False if (value.get("intent") == "permission") != (index is not None): return False phase = value["phase"] intent = value["intent"] role_id = value["roleId"] token_id = value["tokenId"] if phase == "role": return ( intent == "role" and role_id is None and user_id is None and token_id is None and index is None ) if phase == "permissions": return ( intent in {None, "permission"} and role_id is not None and user_id is None and token_id is None ) if phase == "user": return ( intent == "user" and role_id is not None and token_id is None and index is None ) if phase == "membership": return ( intent == "membership" and role_id is not None and user_id is not None and token_id is None and index is None ) if phase == "token": return ( intent == "token" and role_id is not None and user_id is not None and index is None ) return ( phase == "ready" and intent is None and role_id is not None and user_id is not None and token_id is not None and index is None ) def _persist_journal(session: Session, journal: dict[str, Any]) -> bool: if not _valid_journal(journal): return False raw = _journal_bytes(journal) try: set_secret( session, FORGE_READER_BINDING_REF, raw, used_by=["authentik:itworx-forge-controller-reader"], metadata={"managedBy": "appops-forge-controller-reader/v1"}, ) stored = get_secret(session, FORGE_READER_BINDING_REF, touch=False) except Exception: # noqa: BLE001 - never expose database or encrypted payload errors. session.rollback() return False return isinstance(stored, str) and hmac.compare_digest(stored, raw) def _load_or_create_journal( session: Session, context: dict[str, Any], manifest: list[dict[str, str]], ) -> dict[str, Any]: if has_secret(session, FORGE_READER_BINDING_REF): raw = get_secret(session, FORGE_READER_BINDING_REF, touch=False) if not isinstance(raw, str) or not raw: _fail("forge_reader_binding_unreadable") try: journal = json.loads(raw, object_pairs_hook=_unique_object) except (ValueError, TypeError): _fail("forge_reader_binding_invalid") if not _valid_journal(journal): _fail("forge_reader_binding_invalid") if not hmac.compare_digest(raw, _journal_bytes(journal)): _fail("forge_reader_binding_not_canonical") if journal["resourceContext"] != context or journal["permissionManifest"] != manifest: _fail("forge_reader_binding_resource_mismatch") return journal if has_secret(session, FORGE_READER_TOKEN_REF): _fail("forge_reader_token_without_binding") journal = { "version": 1, "nonce": str(uuid4()), "resourceContext": context, "permissionManifest": manifest, "userId": None, "roleId": None, "tokenId": None, "phase": "role", "intent": "role", "permissionIndex": None, } if not _persist_journal(session, journal): _fail("forge_reader_binding_persist_failed", status="partial_failure") return journal def _persist_token(session: Session, token: str) -> bool: try: set_secret( session, FORGE_READER_TOKEN_REF, token, used_by=["authentik:itworx-forge-controller-reader"], metadata={"managedBy": "appops-forge-controller-reader/v1"}, ) stored = get_secret(session, FORGE_READER_TOKEN_REF, touch=False) except Exception: # noqa: BLE001 - token material stays out of every error channel. session.rollback() return False return isinstance(stored, str) and hmac.compare_digest(stored, token) def _reader_secret(session: Session) -> str | None: if not has_secret(session, FORGE_READER_TOKEN_REF): return None token = get_secret(session, FORGE_READER_TOKEN_REF, touch=False) if not isinstance(token, str) or not _TOKEN_PATTERN.fullmatch(token): _fail("forge_reader_token_unreadable") return token def _oidc_resource_context(session: Session, connector: Any) -> tuple[dict[str, Any], str]: if not has_secret(session, FORGE_OIDC_BINDING_SECRET_REF): _fail("forge_reader_oidc_binding_required", status="blocked") if not _apply_forge_oidc_binding_to_runtime(session): _fail("forge_reader_oidc_binding_invalid") raw_binding = get_secret(session, FORGE_OIDC_BINDING_SECRET_REF, touch=False) try: decoded_binding = json.loads(raw_binding, object_pairs_hook=_unique_object) binding = ForgeOidcBindingV1.model_validate(decoded_binding) except Exception: # noqa: BLE001 - stored binding is untrusted. _fail("forge_reader_oidc_binding_invalid") provider_id = binding.provider_id if not isinstance(provider_id, str) or _positive_id(provider_id, provider=True) is None: _fail("forge_reader_oidc_provider_binding_required", status="blocked") if settings.apis.get("AUTHENTIK_FORGE_OIDC_PROVIDER_ID") != provider_id: _fail("forge_reader_oidc_provider_binding_mismatch") oidc_secret = get_secret(session, FORGE_OIDC_SECRET_REF, touch=False) if not isinstance(oidc_secret, str) or not 32 <= len(oidc_secret) <= 255 or not oidc_secret.isascii() or not oidc_secret.isprintable(): _fail("forge_reader_oidc_client_secret_unavailable", status="conflict") config = connector._config() if ( config is None or str(getattr(config, "url", "")).rstrip("/") != _AUTHENTIK_BASE or getattr(config, "verify_tls", None) is not True ): _fail("forge_reader_authentik_configuration_required", status="blocked") maps = { "openid": str(binding.scope_mapping_ids.openid), "email": str(binding.scope_mapping_ids.email), "profile": str(binding.scope_mapping_ids.profile), } key_id = str(settings.apis.get("AUTHENTIK_FORGE_OIDC_SIGNING_KEY_ID") or "") if _uuid(key_id) is None: _fail("forge_reader_signing_key_binding_invalid", status="blocked") applications = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/applications/", {"page_size": 100}, ) app_matches = [ item for item in applications if item.get("slug") == FORGE_OIDC_CLIENT_ID or item.get("name") == FORGE_OIDC_APPLICATION_NAME ] if len(app_matches) != 1: _fail("forge_reader_oidc_application_missing_or_ambiguous") application_id = _uuid(_remote_id(app_matches[0])) if application_id is None: _fail("forge_reader_oidc_application_id_invalid") application = _admin_json(connector, "GET", f"/core/applications/{FORGE_OIDC_CLIENT_ID}/") if ( not isinstance(application, dict) or _uuid(_remote_id(application)) != application_id or application.get("name") != FORGE_OIDC_APPLICATION_NAME or application.get("slug") != FORGE_OIDC_CLIENT_ID or str(application.get("meta_launch_url") or application.get("launch_url") or "").rstrip("/") != FORGE_OIDC_LAUNCH_URL or application.get("meta_description") != FORGE_OIDC_MANAGED_DESCRIPTION or _positive_id(_remote_id(application.get("provider")), provider=True) != provider_id ): _fail("forge_reader_oidc_application_contract_mismatch") application = None provider = _admin_json(connector, "GET", f"/providers/oauth2/{quote(provider_id, safe='')}/") if not isinstance(provider, dict) or _positive_id(_remote_id(provider), provider=True) != provider_id: _fail("forge_reader_oidc_provider_contract_mismatch") if not AuthentikConnector._forge_oidc_provider_matches(provider, signing_key_id=key_id): provider = None _fail("forge_reader_oidc_provider_contract_mismatch") actual_client_secret = provider.get("client_secret") secret_matches = ( isinstance(actual_client_secret, str) and actual_client_secret.isascii() and oidc_secret.isascii() and hmac.compare_digest(actual_client_secret, oidc_secret) ) actual_client_secret = None provider = None if secret_matches is not True: _fail("forge_reader_oidc_client_secret_mismatch") for scope, mapping_id in maps.items(): mapping = _admin_json( connector, "GET", f"/propertymappings/provider/scope/{quote(mapping_id, safe='')}/", ) if not _forge_oidc_scope_mapping_matches(mapping, mapping_id=mapping_id, scope=scope): _fail("forge_reader_oidc_scope_mapping_mismatch") mapping = None all_groups = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/groups/", _GROUP_LIST_PARAMS, ) group_ids: dict[str, str] = {} for name in FORGE_OIDC_GROUPS: matches = [item for item in all_groups if item.get("name") == name] if len(matches) != 1: _fail("forge_reader_oidc_group_missing_or_ambiguous") group_id = _uuid(_remote_id(matches[0])) if group_id is None: _fail("forge_reader_oidc_group_id_invalid") group = _admin_json( connector, "GET", f"/core/groups/{quote(group_id, safe='')}/", params={ "include_users": "true", "include_children": "true", "include_parents": "true", "include_inherited_roles": "true", }, ) if ( not isinstance(group, dict) or _uuid(_remote_id(group)) != group_id or group.get("name") != name or not _forge_oidc_group_contract_matches(group, require_users=True) or group.get("inherited_roles_obj") != [] ): _fail("forge_reader_oidc_group_contract_mismatch") group_ids[name] = group_id group = None if len(set(group_ids.values())) != 3: _fail("forge_reader_oidc_group_id_collision") signing_key = _admin_json( connector, "GET", f"/crypto/certificatekeypairs/{quote(key_id, safe='')}/", ) if ( not isinstance(signing_key, dict) or _uuid(_remote_id(signing_key)) != key_id or str(signing_key.get("key_type") or "").lower() != "rsa" or signing_key.get("private_key_available") is not True ): _fail("forge_reader_signing_key_contract_mismatch") signing_key = None context = { "applicationId": application_id, "providerId": provider_id, "scopeMappingIds": maps, "groupIds": group_ids, "signingKeyId": key_id, } return context, oidc_secret def _role_list(connector: Any, *, name: str | None = None, role_id: str | None = None) -> list[dict[str, Any]]: if role_id is not None: value = _admin_json(connector, "GET", f"/rbac/roles/{quote(role_id, safe='')}/") return [value] if isinstance(value, dict) else [] params: dict[str, Any] = {"page_size": 100} if name is not None: params["name"] = name return _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/rbac/roles/", params, ) def _role_users(connector: Any, role_id: str) -> list[dict[str, Any]]: return _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/users/", {"roles_by_pk": role_id, "page_size": 100}, ) def _role_object_permissions(connector: Any, role_id: str) -> list[dict[str, Any]]: return _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/rbac/permissions/roles/", {"uuid": role_id, "page_size": 100}, ) def _global_permissions(connector: Any, role_id: str) -> list[dict[str, Any]]: return _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/rbac/permissions/", {"role": role_id, "page_size": 100}, ) def _group_has_role(rows: list[dict[str, Any]], role_id: str) -> bool: for group in rows: direct = group.get("roles") inherited = group.get("inherited_roles_obj") if not isinstance(direct, list) or not isinstance(inherited, list): _fail("forge_reader_group_role_readback_invalid", status="failed") if any(str(_remote_id(item)) == role_id for item in direct): return True if any(str(_remote_id(item)) == role_id for item in inherited): return True return False def _role_object_permission_keys(rows: list[dict[str, Any]]) -> set[tuple[str, str, str, str]]: keys: set[tuple[str, str, str, str]] = set() for row in rows: if not isinstance(row, dict): _fail("forge_reader_role_permission_readback_invalid", status="failed") key = ( str(row.get("app_label") or ""), str(row.get("model") or ""), str(row.get("codename") or ""), str(row.get("object_pk") or ""), ) if not all(key) or key in keys: _fail("forge_reader_role_permission_readback_conflict") keys.add(key) return keys def _manifest_keys(rows: list[dict[str, Any]]) -> set[tuple[str, str, str, str]]: return {(row["app"], row["model"], row["codename"], row["objectPk"]) for row in rows} def _verify_permission_catalog(connector: Any, manifest: list[dict[str, Any]]) -> None: for row in manifest: items = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/rbac/permissions/", { "content_type__app_label": row["app"], "content_type__model": row["model"], "codename": row["codename"], "page_size": 100, }, ) if ( len(items) != 1 or items[0].get("app_label") != row["app"] or items[0].get("model") != row["model"] or items[0].get("codename") != row["codename"] ): _fail("forge_reader_permission_catalog_mismatch") def _journal_store( session: Session, journal: dict[str, Any], *, phase: str | None = None, intent: str | None | object = ..., permission_index: int | None | object = ..., user_id: str | None | object = ..., role_id: str | None | object = ..., token_id: str | None | object = ..., ) -> None: updated = dict(journal) if phase is not None: updated["phase"] = phase if intent is not ...: updated["intent"] = intent if permission_index is not ...: updated["permissionIndex"] = permission_index if user_id is not ...: updated["userId"] = user_id if role_id is not ...: updated["roleId"] = role_id if token_id is not ...: updated["tokenId"] = token_id if not _persist_journal(session, updated): _fail("forge_reader_binding_persist_failed", status="partial_failure") journal.clear() journal.update(updated) def _prove_empty_role(connector: Any, role_id: str) -> None: object_rows = _role_object_permissions(connector, role_id) global_rows = _global_permissions(connector, role_id) users = _role_users(connector, role_id) groups = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/groups/", _GROUP_LIST_PARAMS, ) if object_rows or global_rows or users or _group_has_role(groups, role_id): _fail("forge_reader_existing_role_not_empty") def _resolve_role(session: Session, connector: Any, journal: dict[str, Any]) -> str: role_id = journal["roleId"] if role_id is None: if journal["phase"] != "role" or journal["intent"] != "role": _fail("forge_reader_role_binding_missing") matches = _role_list(connector, name=FORGE_READER_ROLE) matches = [row for row in matches if row.get("name") == FORGE_READER_ROLE] if len(matches) > 1: _fail("forge_reader_role_ambiguous") if matches: role_id = _uuid(_remote_id(matches[0])) if role_id is None: _fail("forge_reader_role_id_invalid") _prove_empty_role(connector, role_id) _journal_store(session, journal, phase="permissions", intent=None, permission_index=None, role_id=role_id) return role_id # The protected role intent is durable before this only remote POST. try: _admin_json(connector, "POST", "/rbac/roles/", payload={"name": FORGE_READER_ROLE}) except _ReaderFailure: pass matches = _role_list(connector, name=FORGE_READER_ROLE) matches = [row for row in matches if row.get("name") == FORGE_READER_ROLE] if len(matches) != 1: _fail("forge_reader_role_creation_unconfirmed", status="partial_failure") role_id = _uuid(_remote_id(matches[0])) if role_id is None: _fail("forge_reader_role_id_invalid", status="partial_failure") _prove_empty_role(connector, role_id) _journal_store(session, journal, phase="permissions", intent=None, permission_index=None, role_id=role_id) return role_id details = _role_list(connector, role_id=role_id) if len(details) != 1 or details[0].get("name") != FORGE_READER_ROLE or _uuid(_remote_id(details[0])) != role_id: _fail("forge_reader_role_binding_mismatch") return role_id def _validate_existing_user( user: dict[str, Any], *, journal: dict[str, Any], role_id: str, ) -> str: user_id = _positive_id(_remote_id(user)) if user_id is None: _fail("forge_reader_user_id_invalid") if journal["userId"] is not None and journal["userId"] != user_id: _fail("forge_reader_user_binding_mismatch") attributes = user.get("attributes") if ( user.get("username") != FORGE_READER_USERNAME or user.get("name") != FORGE_READER_NAME or user.get("type") != "service_account" or user.get("path") != FORGE_READER_PATH or user.get("is_active") is not True or user.get("is_superuser") is not False or attributes != {FORGE_READER_OWNER_ATTRIBUTE: journal["nonce"]} ): _fail("forge_reader_existing_user_conflict") groups = user.get("groups") roles = user.get("roles") if not isinstance(groups, list) or not isinstance(roles, list) or groups: _fail("forge_reader_user_membership_conflict") role_ids = {str(_remote_id(item)) for item in roles} if journal["phase"] == "membership" and journal["intent"] == "membership": allowed_role_states = (set(), {role_id}) elif journal["phase"] in {"token", "ready"}: allowed_role_states = ({role_id},) else: allowed_role_states = (set(),) if role_ids not in allowed_role_states or len(roles) != len(role_ids): _fail("forge_reader_user_role_conflict") return user_id def _user_matches_by_username(connector: Any) -> list[dict[str, Any]]: users = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/users/", {"username": FORGE_READER_USERNAME, "include_groups": "true", "include_roles": "true", "page_size": 100}, ) return [user for user in users if user.get("username") == FORGE_READER_USERNAME] def _ensure_role_permissions( session: Session, connector: Any, journal: dict[str, Any], manifest: list[dict[str, str]], ) -> str: role_id = journal["roleId"] if role_id is None: _fail("forge_reader_role_binding_missing") _verify_permission_catalog(connector, manifest) required = _manifest_keys(manifest) current = _role_object_permission_keys(_role_object_permissions(connector, role_id)) globals_ = _global_permissions(connector, role_id) if globals_: _fail("forge_reader_global_permission_conflict") if not current.issubset(required): _fail("forge_reader_object_permission_conflict") if journal["phase"] not in {"permissions"} and current != required: _fail("forge_reader_permissions_incomplete") if journal["phase"] == "permissions" and journal["intent"] not in {None, "permission"}: _fail("forge_reader_journal_stage_conflict") current_list = current members = _role_users(connector, role_id) if journal["phase"] in {"permissions", "user"} and members: _fail("forge_reader_role_has_other_users") groups = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/groups/", _GROUP_LIST_PARAMS, ) if _group_has_role(groups, role_id): _fail("forge_reader_role_assigned_to_group") for index, row in enumerate(manifest): key = (row["app"], row["model"], row["codename"], row["objectPk"]) if key in current_list: continue if journal["phase"] != "permissions": _journal_store(session, journal, phase="permissions", intent="permission", permission_index=index) else: _journal_store(session, journal, phase="permissions", intent="permission", permission_index=index) model = f"{row['app']}.{row['model']}" try: _admin_json( connector, "POST", f"/rbac/permissions/assigned_by_roles/{quote(role_id, safe='')}/assign/", payload={"permissions": [row["codename"]], "model": model, "object_pk": row["objectPk"]}, ) except _ReaderFailure: # A transport failure is reconciled by GET; this invocation never blindly repeats. pass current = _role_object_permission_keys(_role_object_permissions(connector, role_id)) globals_ = _global_permissions(connector, role_id) if globals_ or not current.issubset(required): _fail("forge_reader_permission_assignment_conflict") if key not in current: _fail("forge_reader_permission_assignment_unconfirmed", status="partial_failure") _journal_store(session, journal, phase="permissions", intent=None, permission_index=None) current_list = current final = _role_object_permission_keys(_role_object_permissions(connector, role_id)) if final != required or _global_permissions(connector, role_id): _fail("forge_reader_permissions_incomplete", status="partial_failure") if journal["phase"] == "permissions": _journal_store(session, journal, phase="user", intent="user", permission_index=None) return role_id def _user_detail(connector: Any, user_id: str) -> dict[str, Any]: value = _admin_json( connector, "GET", f"/core/users/{quote(user_id, safe='')}/", params={"include_groups": "true", "include_roles": "true"}, ) if not isinstance(value, dict): _fail("forge_reader_user_readback_invalid", status="failed") return value def _ensure_user(session: Session, connector: Any, journal: dict[str, Any], role_id: str) -> str: matches = _user_matches_by_username(connector) if len(matches) > 1: _fail("forge_reader_user_ambiguous") if matches: user_id = _validate_existing_user( matches[0], journal=journal, role_id=role_id, ) detail = _user_detail(connector, user_id) observed_id = _validate_existing_user( detail, journal=journal, role_id=role_id, ) if observed_id != user_id: _fail("forge_reader_user_readback_mismatch") if journal["userId"] is None: if journal["phase"] != "user" or journal["intent"] != "user": _fail("forge_reader_unbound_user_conflict") _journal_store(session, journal, phase="membership", intent="membership", user_id=user_id) elif journal["phase"] == "user": _journal_store(session, journal, phase="membership", intent="membership", user_id=user_id) return user_id if journal["userId"] is not None: _fail("forge_reader_bound_user_missing") if journal["phase"] != "user" or journal["intent"] != "user": _fail("forge_reader_user_creation_not_authorized") payload = { "username": FORGE_READER_USERNAME, "name": FORGE_READER_NAME, "type": "service_account", "path": FORGE_READER_PATH, "attributes": {FORGE_READER_OWNER_ATTRIBUTE: journal["nonce"]}, "groups": [], "roles": [], } try: _admin_json(connector, "POST", "/core/users/", payload=payload) except _ReaderFailure: pass matches = _user_matches_by_username(connector) if len(matches) != 1: _fail("forge_reader_user_creation_unconfirmed", status="partial_failure") user_id = _validate_existing_user(matches[0], journal=journal, role_id=role_id) detail = _user_detail(connector, user_id) if _validate_existing_user(detail, journal=journal, role_id=role_id) != user_id: _fail("forge_reader_user_readback_mismatch", status="partial_failure") _journal_store(session, journal, phase="membership", intent="membership", user_id=user_id) return user_id def _users_with_role(connector: Any, role_id: str) -> list[dict[str, Any]]: return _role_users(connector, role_id) def _user_roles(connector: Any, user_id: str) -> list[dict[str, Any]]: return _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/rbac/roles/", {"users": user_id, "inherited": "true", "page_size": 100}, ) def _ensure_membership(session: Session, connector: Any, journal: dict[str, Any], user_id: str, role_id: str) -> None: all_groups = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/groups/", _GROUP_LIST_PARAMS, ) if _group_has_role(all_groups, role_id): _fail("forge_reader_role_assigned_to_group") members = _users_with_role(connector, role_id) member_ids = [_positive_id(_remote_id(item)) for item in members] if any(value is None for value in member_ids) or len(set(member_ids)) != len(member_ids): _fail("forge_reader_role_user_readback_invalid") if member_ids not in ([], [user_id]): _fail("forge_reader_role_has_other_users") detail = _user_detail(connector, user_id) _validate_existing_user( detail, journal=journal, role_id=role_id, ) roles = _user_roles(connector, user_id) role_ids = [_uuid(_remote_id(item)) for item in roles] if any(value is None for value in role_ids) or len(set(role_ids)) != len(role_ids): _fail("forge_reader_inherited_role_readback_invalid") if role_ids not in ([], [role_id]): _fail("forge_reader_inherited_role_conflict") has_role_membership = member_ids == [user_id] user_has_role = role_ids == [role_id] if has_role_membership != user_has_role: _fail("forge_reader_role_membership_readback_conflict") if has_role_membership: if journal["phase"] == "membership": _journal_store(session, journal, phase="token", intent="token") return if journal["phase"] != "membership" or journal["intent"] != "membership": _fail("forge_reader_role_membership_not_authorized") _journal_store(session, journal, phase="membership", intent="membership") try: _admin_json( connector, "POST", f"/rbac/roles/{quote(role_id, safe='')}/add_user/", payload={"pk": int(user_id)}, ) except _ReaderFailure: pass members = _users_with_role(connector, role_id) member_ids = [_positive_id(_remote_id(item)) for item in members] roles = _user_roles(connector, user_id) role_ids = [_uuid(_remote_id(item)) for item in roles] if member_ids != [user_id] or role_ids != [role_id]: _fail("forge_reader_role_membership_unconfirmed", status="partial_failure") _journal_store(session, journal, phase="token", intent="token") def _token_metadata(connector: Any) -> list[dict[str, Any]]: return _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/tokens/", {"identifier": FORGE_READER_TOKEN_ID, "page_size": 100}, ) def _validate_token_metadata(token: dict[str, Any], *, user_id: str, journal: dict[str, Any]) -> str: token_id = _uuid(_remote_id(token)) token_user_id = _positive_id(_remote_id(token.get("user"))) if ( token_id is None or token_user_id != user_id or token.get("identifier") != FORGE_READER_TOKEN_ID or token.get("intent") != "api" or token.get("description") != FORGE_READER_TOKEN_DESCRIPTION or token.get("expiring") is not False ): _fail("forge_reader_token_metadata_conflict") if journal["tokenId"] is not None and journal["tokenId"] != token_id: _fail("forge_reader_token_binding_mismatch") return token_id def _ensure_token( session: Session, connector: Any, journal: dict[str, Any], user_id: str, admin_token: str, ) -> str: existing_secret = _reader_secret(session) rows = _token_metadata(connector) rows = [row for row in rows if row.get("identifier") == FORGE_READER_TOKEN_ID] if len(rows) > 1: _fail("forge_reader_token_ambiguous") if rows: if journal["phase"] not in {"token", "ready"} or journal["intent"] != "token" and journal["phase"] != "ready": _fail("forge_reader_unbound_token_conflict") token_id = _validate_token_metadata(rows[0], user_id=user_id, journal=journal) if journal["tokenId"] is None: _journal_store(session, journal, phase="token", intent="token", token_id=token_id) else: if existing_secret is not None: _fail("forge_reader_secret_without_remote_token") if journal["phase"] != "token" or journal["intent"] != "token": _fail("forge_reader_token_creation_not_authorized") try: _admin_json( connector, "POST", "/core/tokens/", payload={ "identifier": FORGE_READER_TOKEN_ID, "intent": "api", "user": int(user_id), "description": FORGE_READER_TOKEN_DESCRIPTION, "expiring": False, }, ) except _ReaderFailure: pass rows = _token_metadata(connector) rows = [row for row in rows if row.get("identifier") == FORGE_READER_TOKEN_ID] if len(rows) != 1: _fail("forge_reader_token_creation_unconfirmed", status="partial_failure") token_id = _validate_token_metadata(rows[0], user_id=user_id, journal=journal) _journal_store(session, journal, phase="token", intent="token", token_id=token_id) if journal["tokenId"] is None: _fail("forge_reader_token_binding_missing") token = _admin_json( connector, "GET", f"/core/tokens/{quote(FORGE_READER_TOKEN_ID, safe='')}/view_key/", ) raw_key = token.get("key") if isinstance(token, dict) else None if not isinstance(raw_key, str) or not _TOKEN_PATTERN.fullmatch(raw_key): token = None raw_key = None _fail("forge_reader_token_key_invalid", status="partial_failure") if existing_secret is None: if has_secret(session, FORGE_READER_TOKEN_REF): raw_key = None _fail("forge_reader_token_unreadable") stored = _persist_token(session, raw_key) if stored is not True: raw_key = None _fail("forge_reader_token_persist_failed", status="partial_failure") elif not hmac.compare_digest(existing_secret, raw_key): raw_key = None _fail("forge_reader_original_token_mismatch") raw_key = None token = None if existing_secret is None: existing_secret = _reader_secret(session) if not isinstance(existing_secret, str): _fail("forge_reader_token_persist_failed", status="partial_failure") if hmac.compare_digest(existing_secret, admin_token): _fail("forge_reader_administrative_token_reuse") return existing_secret def _b64url_int(value: Any) -> int | None: if not isinstance(value, str) or not value: return None try: padded = value + "=" * (-len(value) % 4) decoded = base64.urlsafe_b64decode(padded.encode("ascii")) return int.from_bytes(decoded, "big") if decoded else None except (ValueError, TypeError, UnicodeError): return None def _verify_public_reader( connector: Any, reader_token: str, oidc_secret: str, context: dict[str, Any], role_id: str, user_id: str, ) -> None: self_response = _reader_json(connector, reader_token, "/core/users/me/") reader = self_response.get("user") if isinstance(self_response, dict) else None if ( not isinstance(reader, dict) or _positive_id(_remote_id(reader)) != user_id or reader.get("username") != FORGE_READER_USERNAME or reader.get("name") != FORGE_READER_NAME or reader.get("is_active") is not True or reader.get("is_superuser") is not False or reader.get("type") != "service_account" or reader.get("system_permissions") != [] or reader.get("groups") != [] ): _fail("forge_reader_self_readback_mismatch") roles = reader.get("roles") if not isinstance(roles, list) or len(roles) != 1 or _uuid(_remote_id(roles[0])) != role_id: _fail("forge_reader_self_role_mismatch") apps = _paged( lambda path, params=None: _reader_json(connector, reader_token, path, params=params), "/core/applications/", {"page_size": 100, "search": FORGE_OIDC_CLIENT_ID}, ) app_matches = [ app for app in apps if app.get("slug") == FORGE_OIDC_CLIENT_ID and _uuid(_remote_id(app)) == context["applicationId"] ] if len(app_matches) != 1: _fail("forge_reader_application_visibility_mismatch") application = _reader_json(connector, reader_token, f"/core/applications/{FORGE_OIDC_CLIENT_ID}/") if ( not isinstance(application, dict) or _uuid(_remote_id(application)) != context["applicationId"] or application.get("name") != FORGE_OIDC_APPLICATION_NAME or application.get("slug") != FORGE_OIDC_CLIENT_ID or str(application.get("meta_launch_url") or application.get("launch_url") or "").rstrip("/") != FORGE_OIDC_LAUNCH_URL or application.get("meta_description") != FORGE_OIDC_MANAGED_DESCRIPTION or _positive_id(_remote_id(application.get("provider")), provider=True) != context["providerId"] ): _fail("forge_reader_application_detail_mismatch") application = None provider = _reader_json( connector, reader_token, f"/providers/oauth2/{quote(context['providerId'], safe='')}/", ) if ( not isinstance(provider, dict) or not AuthentikConnector._forge_oidc_provider_matches(provider, signing_key_id=context["signingKeyId"]) or _positive_id(_remote_id(provider), provider=True) != context["providerId"] ): _fail("forge_reader_provider_visibility_mismatch") actual_secret = provider.get("client_secret") secret_matches = ( isinstance(actual_secret, str) and actual_secret.isascii() and oidc_secret.isascii() and hmac.compare_digest(actual_secret, oidc_secret) ) actual_secret = None provider = None if secret_matches is not True: _fail("forge_reader_provider_secret_readback_mismatch") for scope, mapping_id in context["scopeMappingIds"].items(): mapping = _reader_json( connector, reader_token, f"/propertymappings/provider/scope/{quote(mapping_id, safe='')}/", ) if not _forge_oidc_scope_mapping_matches(mapping, mapping_id=mapping_id, scope=scope): _fail("forge_reader_scope_mapping_visibility_mismatch") visible_groups = _paged( lambda path, params=None: _reader_json(connector, reader_token, path, params=params), "/core/groups/", _GROUP_LIST_PARAMS, ) visible_by_id = {_uuid(_remote_id(group)): group for group in visible_groups} if len(visible_by_id) != 3 or set(visible_by_id) != set(context["groupIds"].values()): _fail("forge_reader_group_visibility_mismatch") for name in FORGE_OIDC_GROUPS: group_id = context["groupIds"][name] group = visible_by_id[group_id] if ( group.get("name") != name or not _forge_oidc_group_contract_matches(group, require_users=True) or group.get("inherited_roles_obj") != [] ): _fail("forge_reader_group_readback_mismatch") key_id = context["signingKeyId"] key = _reader_json( connector, reader_token, f"/crypto/certificatekeypairs/{quote(key_id, safe='')}/", ) if ( not isinstance(key, dict) or _uuid(_remote_id(key)) != key_id or str(key.get("key_type") or "").lower() != "rsa" or key.get("private_key_available") is not True ): _fail("forge_reader_signing_key_readback_mismatch") cert_response = _reader_json( connector, reader_token, f"/crypto/certificatekeypairs/{quote(key_id, safe='')}/view_certificate/", ) certificate_data = cert_response.get("data") if isinstance(cert_response, dict) else None if not isinstance(certificate_data, str): _fail("forge_reader_public_certificate_readback_invalid") try: certificate = x509.load_pem_x509_certificate(certificate_data.encode("ascii")) public_key = certificate.public_key() except (ValueError, TypeError, UnicodeError): _fail("forge_reader_public_certificate_readback_invalid") if not isinstance(public_key, rsa.RSAPublicKey): _fail("forge_reader_signing_key_not_rsa") numbers = public_key.public_numbers() discovery = _public_json(FORGE_OIDC_DISCOVERY_URL) if ( not isinstance(discovery, dict) or discovery.get("issuer") != FORGE_OIDC_ISSUER or "code" not in discovery.get("response_types_supported", []) or "authorization_code" not in discovery.get("grant_types_supported", []) or not {"openid", "email", "profile"}.issubset(set(discovery.get("scopes_supported", []))) or "RS256" not in discovery.get("id_token_signing_alg_values_supported", []) or discovery.get("jwks_uri") != f"{FORGE_OIDC_ISSUER}jwks/" ): _fail("forge_reader_discovery_contract_mismatch") jwks_url = discovery["jwks_uri"] jwks = _public_json(jwks_url) keys = jwks.get("keys") if isinstance(jwks, dict) else None if not isinstance(keys, list): _fail("forge_reader_jwks_invalid") matches = [ item for item in keys if isinstance(item, dict) and item.get("kty") == "RSA" and item.get("alg") == "RS256" and item.get("use") == "sig" and _b64url_int(item.get("n")) == numbers.n and _b64url_int(item.get("e")) == numbers.e ] if len(matches) != 1: _fail("forge_reader_jwks_signing_key_mismatch") certificate_data = None public_key = None certificate = None jwks = None discovery = None @dataclass(frozen=True, slots=True, repr=False) class _VerifiedForgeDeliveryInputs: """Held delivery inputs; deliberately cannot print secret-bearing fields.""" context: dict[str, Any] oidc_client_secret: bytes = field(repr=False) authentik_read_token: bytes = field(repr=False) def _verify_forge_delivery_inputs( session: Session, connector: Any, ) -> _VerifiedForgeDeliveryInputs: """Revalidate the complete fixed Forge reader contract without writes.""" if not has_secret(session, FORGE_OIDC_BINDING_SECRET_REF): _fail("forge_reader_oidc_binding_required", status="blocked") if not has_secret(session, FORGE_READER_BINDING_REF): _fail("forge_reader_binding_required", status="blocked") if not has_secret(session, FORGE_READER_TOKEN_REF): _fail("forge_reader_token_required", status="blocked") raw_journal = get_secret(session, FORGE_READER_BINDING_REF, touch=False) try: journal = json.loads(raw_journal, object_pairs_hook=_unique_object) except (TypeError, ValueError, UnicodeError): _fail("forge_reader_binding_invalid") if not _valid_journal(journal): _fail("forge_reader_binding_invalid") if not hmac.compare_digest(raw_journal, _journal_bytes(journal)): _fail("forge_reader_binding_not_canonical") if ( journal["phase"] != "ready" or journal["intent"] is not None or journal["permissionIndex"] is not None or not journal["userId"] or not journal["roleId"] or not journal["tokenId"] ): _fail("forge_reader_binding_not_ready", status="blocked") context, oidc_secret = _oidc_resource_context(session, connector) manifest = _manifest(context) if journal["resourceContext"] != context or journal["permissionManifest"] != manifest: _fail("forge_reader_binding_resource_mismatch") role_id = journal["roleId"] user_id = journal["userId"] role_rows = _role_list(connector, role_id=role_id) if ( len(role_rows) != 1 or role_rows[0].get("name") != FORGE_READER_ROLE or _uuid(_remote_id(role_rows[0])) != role_id ): _fail("forge_reader_role_binding_mismatch") expected_permissions = _manifest_keys(manifest) actual_permissions = _role_object_permission_keys( _role_object_permissions(connector, role_id) ) if actual_permissions != expected_permissions: _fail("forge_reader_permissions_incomplete") if _global_permissions(connector, role_id): _fail("forge_reader_global_permission_conflict") members = _role_users(connector, role_id) member_ids = [_positive_id(_remote_id(item)) for item in members] if member_ids != [user_id]: _fail("forge_reader_role_membership_readback_conflict") all_groups = _paged( lambda path, params=None: _admin_json(connector, "GET", path, params=params), "/core/groups/", _GROUP_LIST_PARAMS, ) if _group_has_role(all_groups, role_id): _fail("forge_reader_role_assigned_to_group") users = _user_matches_by_username(connector) if len(users) != 1: _fail("forge_reader_user_ambiguous") user = _user_detail(connector, user_id) if _validate_existing_user(user, journal=journal, role_id=role_id) != user_id: _fail("forge_reader_user_readback_mismatch") if _positive_id(_remote_id(users[0])) != user_id: _fail("forge_reader_user_binding_mismatch") user_roles = _user_roles(connector, user_id) user_role_ids = [_uuid(_remote_id(item)) for item in user_roles] if user_role_ids != [role_id]: _fail("forge_reader_user_role_conflict") tokens = [ item for item in _token_metadata(connector) if item.get("identifier") == FORGE_READER_TOKEN_ID ] if len(tokens) != 1: _fail("forge_reader_token_ambiguous") if _validate_token_metadata(tokens[0], user_id=user_id, journal=journal) != journal["tokenId"]: _fail("forge_reader_token_binding_mismatch") reader_token = _reader_secret(session) if reader_token is None: _fail("forge_reader_token_required", status="blocked") admin_config = connector._config() admin_token = getattr(admin_config, "token", None) if admin_config is not None else None if isinstance(admin_token, str) and hmac.compare_digest(reader_token, admin_token): _fail("forge_reader_administrative_token_reuse") if ( not isinstance(oidc_secret, str) or not oidc_secret.isascii() or not oidc_secret.isprintable() or not reader_token.isascii() ): _fail("forge_reader_delivery_input_invalid") _verify_public_reader( connector, reader_token, oidc_secret, context, role_id, user_id, ) return _VerifiedForgeDeliveryInputs( context=dict(context), oidc_client_secret=oidc_secret.encode("ascii"), authentik_read_token=reader_token.encode("ascii"), ) def _preflight_collisions( connector: Any, journal: dict[str, Any], role_id: str | None, existing_token: str | None, ) -> None: users = _user_matches_by_username(connector) if len(users) > 1: _fail("forge_reader_user_ambiguous") if users: if role_id is None: role_id = journal.get("roleId") if role_id is None: _fail("forge_reader_existing_user_conflict") _validate_existing_user( users[0], journal=journal, role_id=role_id, ) token_rows = _token_metadata(connector) token_rows = [row for row in token_rows if row.get("identifier") == FORGE_READER_TOKEN_ID] if len(token_rows) > 1: _fail("forge_reader_token_ambiguous") if token_rows and ( journal["phase"] not in {"token", "ready"} or (journal["intent"] != "token" and journal["phase"] != "ready") ): _fail("forge_reader_unbound_token_conflict") if token_rows: if journal["userId"] is None: _fail("forge_reader_unbound_token_conflict") if len(token_rows) != 1: _fail("forge_reader_token_ambiguous") token_id = _validate_token_metadata(token_rows[0], user_id=journal["userId"], journal=journal) if journal["tokenId"] is not None and journal["tokenId"] != token_id: _fail("forge_reader_token_binding_mismatch") elif existing_token is not None or journal["tokenId"] is not None: _fail("forge_reader_secret_without_remote_token") def _provision_locked(session: Session, connector: Any) -> dict[str, Any]: context, oidc_secret = _oidc_resource_context(session, connector) manifest = _manifest(context) journal = _load_or_create_journal(session, context, manifest) existing_token = _reader_secret(session) role_id = journal["roleId"] _preflight_collisions(connector, journal, role_id, existing_token) role_id = _resolve_role(session, connector, journal) _ensure_role_permissions(session, connector, journal, manifest) user_id = _ensure_user(session, connector, journal, role_id) _ensure_membership(session, connector, journal, user_id, role_id) admin_token = str(connector._config().token) reader_token = _ensure_token(session, connector, journal, user_id, admin_token) _verify_public_reader(connector, reader_token, oidc_secret, context, role_id, user_id) _journal_store(session, journal, phase="ready", intent=None, permission_index=None, token_id=journal["tokenId"]) return _result("ready", "forge_reader_ready", ready=True) def _result(status: str, message: str, *, ready: bool = False) -> dict[str, Any]: return { "success": ready, "status": status, "message": message, "ready": ready, "secretRef": FORGE_READER_TOKEN_REF, } def _audit_result(session: Session, identity: str, outcome: dict[str, Any]) -> bool: try: log_event( session, source="manual", identity=identity, action_id=FORGE_OIDC_ACTION, target=FORGE_OIDC_TARGET, target_type="authentik_service_account", risk="critical", policy_decision="allowed" if outcome["status"] != "blocked" else "blocked", reason=outcome["message"], result=outcome["status"], redacted_metadata={"reader": FORGE_READER_USERNAME, "secretRef": FORGE_READER_TOKEN_REF}, ) return True except Exception: # noqa: BLE001 - audit failures cannot expose credentials or remote bodies. session.rollback() return False def provision_forge_controller_reader( session: Session, *, actor_identity: str, authentik: Any | None = None, ) -> dict[str, Any]: """Provision and verify the one fixed, read-only Forge Authentik identity.""" identity = actor_identity.strip() if isinstance(actor_identity, str) else "" if not identity: outcome = _result("blocked", "forge_reader_operator_identity_required") else: try: with _forge_oidc_process_lock(): if not _exact_policy_allowed(session): outcome = _result("blocked", "forge_reader_exact_target_policy_required") elif secret_storage_status().get("productionReady") is not True: outcome = _result("blocked", "forge_reader_secret_store_not_production_ready") else: if authentik is None: from .setup_service import apply_stored_connector_settings apply_stored_connector_settings(session, {"authentik"}) outcome = _provision_locked(session, authentik or AuthentikConnector()) except _ReaderFailure as exc: outcome = _result(exc.status, exc.code) except Exception: # noqa: BLE001 - fail closed without remote details or credentials. outcome = _result("failed", "forge_reader_provisioning_failed") if not _audit_result(session, identity or "system", outcome) and outcome["ready"] is True: return _result("partial_failure", "forge_reader_audit_persist_failed") return outcome