diff --git a/.github/workflows/foundation-ci.yml b/.github/workflows/foundation-ci.yml index 6b475d6f..4f6d0536 100644 --- a/.github/workflows/foundation-ci.yml +++ b/.github/workflows/foundation-ci.yml @@ -68,6 +68,7 @@ jobs: PYTHONPATH=packages/selection-review/src COVERAGE_FILE=/tmp/orgmetra-selection-review.coverage python -m pytest -c packages/selection-review/pyproject.toml packages/selection-review/tests PYTHONPATH=services/job-analysis-api/src:packages/hris-kernel/src:packages/keyverse-adapter/src COVERAGE_FILE=/tmp/orgmetra-job-analysis-api.coverage python -m pytest -c services/job-analysis-api/pyproject.toml services/job-analysis-api/tests PYTHONPATH=services/people-api/src:packages/hris-kernel/src:packages/keyverse-adapter/src COVERAGE_FILE=/tmp/orgmetra-people-api.coverage python -m pytest -c services/people-api/pyproject.toml services/people-api/tests + PYTHONPATH=services/workforce-validation-api/src:packages/keyverse-adapter/src COVERAGE_FILE=/tmp/orgmetra-workforce-validation-api.coverage python -m pytest -c services/workforce-validation-api/pyproject.toml services/workforce-validation-api/tests - name: Run PostgreSQL contracts in isolated containers env: PGPASSWORD: orgmetra @@ -83,6 +84,7 @@ jobs: test_audit_outbox_hardening_postgres.sh test_candidate_worker_conversion_postgres.sh test_validity_study_case_postgres.sh + test_workforce_validation_owner_schema_postgres.sh test_criterion_observation_scope_postgres.sh test_people_mutation_idempotency_postgres.sh test_job_analysis_snapshot_postgres.sh diff --git a/manifest.json b/manifest.json index f7b6cf55..f4085d0e 100644 --- a/manifest.json +++ b/manifest.json @@ -5,9 +5,9 @@ "files": [ { "path": ".github/workflows/foundation-ci.yml", - "sha256": "b6a4365936b66803a8112f034c77d53d33301a7a798ed4f68746a4f2d8b081d7", - "bytes": 6651, - "lines": 125 + "sha256": "31c6a46cb81513cdaa2a08eed5cb00a57d8b36f15e3c772d230452cdc3329aed", + "bytes": 6974, + "lines": 127 }, { "path": ".gitignore", @@ -359,9 +359,9 @@ }, { "path": "scripts/foundation-contract-core.mjs", - "sha256": "9b03efbbdffa60a05f5924e8a61b1cbc3cd75c502df428a5920085e8d0bf3603", - "bytes": 28121, - "lines": 688 + "sha256": "5dfc54d40820dfc45962dcc91367c57baf6b011e68efc5f6e8d59e7e82145b2a", + "bytes": 28182, + "lines": 689 }, { "path": "scripts/foundation-contract.mjs", @@ -465,11 +465,17 @@ "bytes": 14708, "lines": 301 }, + { + "path": "tests/test_workforce_validation_owner_schema_postgres.sh", + "sha256": "29f0cd8a7d9040ff86095b68fa2f3d0ed54ea3777e5a816d4a79eb6ceafb9339", + "bytes": 2949, + "lines": 76 + }, { "path": "tests/validate_repository.py", - "sha256": "091836b2f68600a30b08f7da2cea8b3bef10201a123da720a7369bf10985eec2", - "bytes": 27237, - "lines": 637 + "sha256": "244627252e7392e4dbed98392c132cb86dc8e5ead839a1129298e8d022fcf8eb", + "bytes": 27300, + "lines": 638 } ] } diff --git a/scripts/foundation-contract-core.mjs b/scripts/foundation-contract-core.mjs index 4aacefb3..ba2d8c00 100644 --- a/scripts/foundation-contract-core.mjs +++ b/scripts/foundation-contract-core.mjs @@ -85,6 +85,7 @@ export const REQUIRED_FILES = Object.freeze([ 'tests/test_audit_outbox_hardening_postgres.sh', 'tests/test_candidate_worker_conversion_postgres.sh', 'tests/test_validity_study_case_postgres.sh', + 'tests/test_workforce_validation_owner_schema_postgres.sh', 'tests/test_criterion_observation_scope_postgres.sh', 'tests/test_people_mutation_idempotency_postgres.sh', 'tests/test_job_analysis_snapshot_postgres.sh', @@ -685,4 +686,4 @@ export function runCli(rootPath, outputStream = process.stdout, errorStream = pr } errorStream.write(`${JSON.stringify({ status: 'failed', error_count: errors.length, errors }, null, 2)}\n`); return 1; -} +} \ No newline at end of file diff --git a/services/workforce-validation-api/README.md b/services/workforce-validation-api/README.md new file mode 100644 index 00000000..d3d84736 --- /dev/null +++ b/services/workforce-validation-api/README.md @@ -0,0 +1,44 @@ +# Orgmetra Workforce Validation API + +This package is the application boundary for the `workforce_validation` bounded context. The current slice exposes one purpose-bound read use case for the existing validity-study registry header and establishes the context-local PostgreSQL ownership bootstrap. + +It does **not** query People, Talent Acquisition, Performance Management, Job Architecture, Psychometrics Commons, fast-mlsirm, or TEPP tables. Those contexts remain separate owners. Exact foreign identifiers and immutable specialist result references cross the boundary only through published contracts. + +## Current slice + +`read_validity_study(...)`: + +- accepts structurally immutable authenticated Keyverse identity attributes, not credentials; +- reconstructs and revalidates principal storage before building the access request, so exact tuple type alone is not treated as identity authority; +- requires both exact `UUID` outer type and exact built-in integer UUID payload before any sentinel/range comparison, so a forged exact UUID with executable internal storage is rejected without invoking caller-defined equality behavior; +- stores UUID identity evidence behind the tuple-backed principal/record/view as exact integer payloads and reconstructs fresh UUID objects at public boundaries, so a retained UUID reference cannot rewrite accepted tenant/study/criterion identity through `object.__setattr__`; +- preserves tenant/study authorization targets as immutable integer snapshots across the executable repository call, so a repository cannot make a foreign record self-consistent by mutating the UUID objects it receives; +- inertly verifies that the owner repository exposes a statically callable `read_validity_study` capability before authorization, without executing caller-controlled descriptors; +- evaluates tenant, purpose, operation, scope, resource, and requested fields before persistence; +- calls only a `ValidityStudyReadPort` owned by this context; +- reconstructs persisted registry scalars into structurally immutable owner evidence before target validation and output; +- returns only the fields authorized for the exact study record; UUID-valued projected fields are reconstituted fresh rather than exposing mutable internal UUID aliases; +- issues `ValidityStudyView` only from the authorized read path. Its public constructor fails closed, and the returned tuple-backed projection cannot be rewritten through ordinary assignment or `object.__setattr__`. + +`ValidityStudyView` is a data projection, not a durable authorization credential or cryptographic capability. Downstream consequential actions must perform their own purpose-bound authorization and authoritative re-resolution rather than treating the Python runtime type as reusable authority. Low-level interpreter construction is outside the supported public API and is not accepted as proof that authorization occurred. + +`services/workforce-validation-api/database/migrations/0001_owner_schema.sql` starts this bounded context's own migration history. It creates the `workforce_validation` schema and deny-default `workforce_validation_role`, revokes public schema access, and intentionally creates or moves no application table yet. The role is a **NOLOGIN migration/schema owner only**; runtime principals must not be granted that owner role. PostgreSQL applies role-level configuration defaults at login and does not re-apply them on `SET ROLE`, so an `ALTER ROLE ... SET search_path` entry on this NOLOGIN role is not treated as a runtime isolation control. The later durable adapter must use a distinct least-privilege runtime role, schema-qualified `workforce_validation` relations, and explicit function-level `search_path` where `SECURITY DEFINER` code is introduced. + +Protected foundation migrations still create validity-study tables in the legacy foundation schema, so the next forward-only persistence increment must adopt those records without normalizing `public.validity_study` as a long-lived service contract or breaking existing linkage evidence. + +Issue #234 owns the remaining order: durable owner-schema adoption and PostgreSQL adapter, idempotent registration, explicit predictor/sample/decision-policy/analysis-protocol versions, scientific adapters, OpenAPI/gateway exposure, and realistic p95 measurement. Issues #236–#244 retain the current bootstrap trust-boundary findings through exact-head acceptance and protected integration: persisted-record immutability, principal immutability and constructor revalidation, owner-role/runtime-role separation, inert repository-capability validation, immutable minimized output, non-public issuance of that output, detached UUID storage/target snapshots, and exact validation of UUID internal payloads before comparison. + +## Test + +The Draft branch is admitted to the canonical Foundation quality workflow with the same hash-locked test toolchain and direct source-tree dependency policy used by the existing owner services: + +```bash +PYTHONPATH=services/workforce-validation-api/src:packages/keyverse-adapter/src \ + COVERAGE_FILE=/tmp/orgmetra-workforce-validation-api.coverage \ + python -m pytest -c services/workforce-validation-api/pyproject.toml \ + services/workforce-validation-api/tests +``` + +The same Foundation job also runs `tests/test_workforce_validation_owner_schema_postgres.sh` in its own pinned PostgreSQL 16.14 container. That contract executes the service-local owner migration and checks the exact deny-default role flags, schema owner, absence of ineffective login-only `rolconfig`, actual `SET ROLE` search-path behavior, absence of inherited PUBLIC `USAGE`/`CREATE`, and absence of application relations in the bootstrap schema. The test intentionally demonstrates that `SET ROLE` retains the caller's existing `search_path`; runtime isolation therefore cannot be inferred from owner-role metadata. + +Those source contracts are not terminal acceptance by themselves. The slice remains Draft until the exact current head actually executes with 100% owned statement/branch coverage, the PostgreSQL owner-schema contract is GREEN, applicable security workflows are terminal, and the normal review/governance requirements are satisfied. Only then may the next forward-only owner-table adoption and durable adapter be treated as eligible for integration. diff --git a/services/workforce-validation-api/database/migrations/0001_owner_schema.sql b/services/workforce-validation-api/database/migrations/0001_owner_schema.sql new file mode 100644 index 00000000..67a1e3c3 --- /dev/null +++ b/services/workforce-validation-api/database/migrations/0001_owner_schema.sql @@ -0,0 +1,27 @@ +-- Establish the logical PostgreSQL ownership boundary for workforce_validation. +-- This migration intentionally creates no application table. Legacy foundation +-- validity-study tables stay untouched until an explicit forward-only adoption +-- migration can preserve existing foreign-key and acceptance contracts. + +BEGIN; + +CREATE ROLE workforce_validation_role NOLOGIN + NOSUPERUSER + NOCREATEDB + NOCREATEROLE + NOINHERIT + NOREPLICATION + NOBYPASSRLS; + +CREATE SCHEMA workforce_validation AUTHORIZATION workforce_validation_role; +REVOKE ALL ON SCHEMA workforce_validation FROM PUBLIC; + +-- workforce_validation_role is a migration/schema-owner identity only. Runtime +-- principals must not be granted this owner role. PostgreSQL role-level GUC +-- defaults apply at login and are not re-applied by SET ROLE; because this role +-- is NOLOGIN, an ALTER ROLE ... SET search_path entry would not provide runtime +-- isolation. Future runtime adapters must use a distinct least-privilege role, +-- schema-qualified owner relations, and explicit function-level search_path for +-- any SECURITY DEFINER code. + +COMMIT; diff --git a/services/workforce-validation-api/pyproject.toml b/services/workforce-validation-api/pyproject.toml new file mode 100644 index 00000000..9a84581c --- /dev/null +++ b/services/workforce-validation-api/pyproject.toml @@ -0,0 +1,41 @@ +[build-system] +requires = ["setuptools==82.0.1"] +build-backend = "setuptools.build_meta" + +[project] +name = "orgmetra-workforce-validation-api" +version = "0.1.0" +description = "Purpose-bound owner boundary for Orgmetra workforce-validation studies." +readme = "README.md" +requires-python = ">=3.11" +license = { text = "Apache-2.0" } +authors = [{ name = "ContextualWisdomLab" }] +dependencies = [ + "orgmetra-keyverse-adapter==0.1.0", +] + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +orgmetra_workforce_validation_api = ["py.typed"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = [ + "--cov=orgmetra_workforce_validation_api", + "--cov-branch", + "--cov-report=term-missing", + "--cov-fail-under=100", +] + +[tool.coverage.run] +branch = true +source = ["orgmetra_workforce_validation_api"] + +[tool.coverage.report] +fail_under = 100 +show_missing = true diff --git a/services/workforce-validation-api/src/orgmetra_workforce_validation_api/__init__.py b/services/workforce-validation-api/src/orgmetra_workforce_validation_api/__init__.py new file mode 100644 index 00000000..48570a97 --- /dev/null +++ b/services/workforce-validation-api/src/orgmetra_workforce_validation_api/__init__.py @@ -0,0 +1,21 @@ +"""Canonical workforce-validation application contracts for Orgmetra.""" + +from orgmetra_workforce_validation_api.registry import ( + ValidationPrincipal, + ValidityStudyIntegrityError, + ValidityStudyNotFound, + ValidityStudyReadPort, + ValidityStudyRecord, + ValidityStudyView, + read_validity_study, +) + +__all__ = [ + "ValidationPrincipal", + "ValidityStudyIntegrityError", + "ValidityStudyNotFound", + "ValidityStudyReadPort", + "ValidityStudyRecord", + "ValidityStudyView", + "read_validity_study", +] diff --git a/services/workforce-validation-api/src/orgmetra_workforce_validation_api/py.typed b/services/workforce-validation-api/src/orgmetra_workforce_validation_api/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/services/workforce-validation-api/src/orgmetra_workforce_validation_api/registry.py b/services/workforce-validation-api/src/orgmetra_workforce_validation_api/registry.py new file mode 100644 index 00000000..10159e1d --- /dev/null +++ b/services/workforce-validation-api/src/orgmetra_workforce_validation_api/registry.py @@ -0,0 +1,451 @@ +"""Purpose-bound application boundary for the workforce-validation study registry. + +This module deliberately stops before PostgreSQL. The protected foundation still +stores validity-study tables in the legacy foundation schema, while +``ARCHITECTURE.md`` assigns persistence ownership to ``workforce_validation``. +The application contract therefore depends on an owner repository port instead +of normalizing direct cross-context SQL into a long-lived service contract. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from inspect import getattr_static +import re +from types import FunctionType +from typing import Protocol, runtime_checkable +from uuid import UUID +from zoneinfo import ZoneInfo + +from orgmetra_keyverse_adapter import ( + PurposeBoundAccessPolicy, + PurposeBoundAccessRequest, + require_purpose_bound_access, +) + +_MAX_UUID_INT = (1 << 128) - 1 +_CODE_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$") +_REFERENCE_PATTERN = re.compile(r"^[a-z][a-z0-9_]*:[A-Za-z0-9][A-Za-z0-9._~-]*$") +_SCOPE_PATTERN = re.compile(r"^orgmetra(?:\.[a-z][a-z0-9_]*){2,}$") +_RESOURCE_KIND = "validity_study_record" +_OPERATION = "read" +_READ_FIELDS = frozenset( + { + "criterion_blueprint_id", + "study_status_code", + "recorded_from", + "recorded_to", + } +) + + +class ValidityStudyNotFound(LookupError): + """Indicate that an authorized study identity has no visible registry record.""" + + +class ValidityStudyIntegrityError(RuntimeError): + """Indicate that persistence returned a record outside the authorized target.""" + + +def _require_operational_uuid(field_name: str, value: object) -> int: + """Return one inert UUID integer after exact outer and internal-type validation.""" + if type(value) is not UUID: + raise ValueError(f"{field_name} must be an exact operational UUID.") + identity = value.int + if type(identity) is not int or identity <= 0 or identity >= _MAX_UUID_INT: + raise ValueError(f"{field_name} must be an exact operational UUID.") + return identity + + +def _store_operational_uuid(field_name: str, value: object) -> int: + """Reduce one validated UUID to immutable integer storage without retaining its object alias.""" + return _require_operational_uuid(field_name, value) + + +def _restore_operational_uuid(field_name: str, value: object) -> UUID: + """Reconstruct one fresh UUID from immutable internal integer storage.""" + if type(value) is not int or value <= 0 or value >= _MAX_UUID_INT: + raise ValueError(f"{field_name} must be an exact operational UUID.") + return UUID(int=value) + + +def _require_code(field_name: str, value: object) -> str: + """Return one exact lower-snake-case code used in an auditable policy request.""" + if type(value) is not str or _CODE_PATTERN.fullmatch(value) is None: + raise ValueError(f"{field_name} must be an exact lower snake_case code.") + return value + + +def _require_aware_datetime(field_name: str, value: object) -> datetime: + """Detach one durable timestamp to exact UTC without arbitrary timezone callbacks.""" + if type(value) is not datetime: + raise ValueError(f"{field_name} must be an exact datetime.") + provider = value.tzinfo + if type(provider) is not timezone and type(provider) is not ZoneInfo: + raise ValueError(f"{field_name} must use a standard-library timezone provider.") + return value.astimezone(timezone.utc) + + +def _validate_scope_set(values: object) -> frozenset[str]: + """Require immutable explicit Keyverse scopes before constructing an access request.""" + if type(values) is not frozenset or not values: + raise ValueError("granted_scope_codes must be a non-empty exact frozenset.") + if any(type(value) is not str or _SCOPE_PATTERN.fullmatch(value) is None for value in values): + raise ValueError("granted_scope_codes must contain exact Orgmetra scopes.") + return values + + +def _validate_requested_fields(values: object) -> frozenset[str]: + """Require a non-empty immutable subset of the published registry read fields.""" + if type(values) is not frozenset or not values: + raise ValueError("requested_fields must be a non-empty exact frozenset.") + if any(type(value) is not str for value in values) or not values.issubset(_READ_FIELDS): + raise ValueError("requested_fields contains a field outside the validity-study registry contract.") + return values + + +def _detach_policy(policy: PurposeBoundAccessPolicy) -> PurposeBoundAccessPolicy: + """Copy policy evidence into exact inert values before any authorization comparison. + + The protected Keyverse adapter accepts subclass-compatible scalar inputs for + backward compatibility. This owner boundary is stricter because a caller- + defined ``str``/``UUID`` subtype could otherwise execute Python behavior when + the evaluator compares or hashes policy attributes. Immutable UUID integer + storage also prevents a retained policy UUID alias from switching the tenant + after this boundary has accepted it. + """ + tenant_record_id = policy.tenant_record_id + policy_version_code = policy.policy_version_code + resource_kind = policy.resource_kind + purpose_code = policy.purpose_code + operation_code = policy.operation_code + required_scope_code = policy.required_scope_code + permitted_fields = policy.permitted_fields + + tenant_identity = _store_operational_uuid("policy tenant_record_id", tenant_record_id) + for field_name, value in ( + ("policy_version_code", policy_version_code), + ("resource_kind", resource_kind), + ("purpose_code", purpose_code), + ("operation_code", operation_code), + ("required_scope_code", required_scope_code), + ): + if type(value) is not str: + raise ValueError(f"policy {field_name} must be an exact string.") + if type(permitted_fields) is not frozenset or any( + type(value) is not str for value in permitted_fields + ): + raise ValueError("policy permitted_fields must contain exact strings in an exact frozenset.") + + return PurposeBoundAccessPolicy( + tenant_record_id=_restore_operational_uuid("policy tenant_record_id", tenant_identity), + policy_version_code=policy_version_code, + resource_kind=resource_kind, + purpose_code=purpose_code, + operation_code=operation_code, + required_scope_code=required_scope_code, + permitted_fields=permitted_fields, + ) + + +class ValidationPrincipal(tuple): + """Structurally immutable authenticated Keyverse attributes for validation reads. + + The bearer credential itself never enters this value. Tuple-backed storage + keeps only immutable UUID integer evidence plus immutable actor/scope values, + so retained UUID references cannot rewrite tenant identity after validation. + """ + + __slots__ = () + + def __new__( + cls, + *, + tenant_record_id: UUID, + actor_reference: str, + granted_scope_codes: frozenset[str], + ) -> ValidationPrincipal: + """Validate exact identity evidence before creating the immutable principal.""" + tenant_identity = _store_operational_uuid("tenant_record_id", tenant_record_id) + if type(actor_reference) is not str or _REFERENCE_PATTERN.fullmatch(actor_reference) is None: + raise ValueError("actor_reference must be an exact namespaced opaque reference.") + scope_codes = _validate_scope_set(granted_scope_codes) + return tuple.__new__(cls, (tenant_identity, actor_reference, scope_codes)) + + @property + def tenant_record_id(self) -> UUID: + """Return a fresh authenticated tenant identity.""" + return _restore_operational_uuid("tenant_record_id", self[0]) + + @property + def actor_reference(self) -> str: + """Return the opaque authenticated actor reference.""" + return self[1] + + @property + def granted_scope_codes(self) -> frozenset[str]: + """Return the immutable authenticated scope set.""" + return self[2] + + +class ValidityStudyRecord(tuple): + """Structurally immutable owner projection of one recorded validity-study header. + + The tuple-backed representation stores UUIDs as immutable integers, preventing + a repository adapter that retains accepted UUID objects from rewriting durable + study identity through ``object.__setattr__`` after construction. Only fields + already represented by the protected foundation schema are carried here. + Predictor, sample, decision-policy and analysis-protocol versions remain a + later scientific-model increment owned by Issue #234. + """ + + __slots__ = () + + def __new__( + cls, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + criterion_blueprint_id: UUID, + study_status_code: str, + recorded_from: datetime, + recorded_to: datetime | None, + ) -> ValidityStudyRecord: + """Validate and detach durable scalars before creating the immutable tuple.""" + tenant_identity = _store_operational_uuid("tenant_record_id", tenant_record_id) + study_identity = _store_operational_uuid("validity_study_id", validity_study_id) + criterion_identity = _store_operational_uuid( + "criterion_blueprint_id", criterion_blueprint_id + ) + status_code = _require_code("study_status_code", study_status_code) + recorded_start = _require_aware_datetime("recorded_from", recorded_from) + recorded_end = ( + None + if recorded_to is None + else _require_aware_datetime("recorded_to", recorded_to) + ) + if recorded_end is not None and recorded_end <= recorded_start: + raise ValueError("recorded_to must be later than recorded_from.") + return tuple.__new__( + cls, + ( + tenant_identity, + study_identity, + criterion_identity, + status_code, + recorded_start, + recorded_end, + ), + ) + + @property + def tenant_record_id(self) -> UUID: + """Return a fresh tenant identity for this validity study.""" + return _restore_operational_uuid("tenant_record_id", self[0]) + + @property + def validity_study_id(self) -> UUID: + """Return a fresh stable validity-study identity.""" + return _restore_operational_uuid("validity_study_id", self[1]) + + @property + def criterion_blueprint_id(self) -> UUID: + """Return a fresh criterion-blueprint identity linked to the study header.""" + return _restore_operational_uuid("criterion_blueprint_id", self[2]) + + @property + def study_status_code(self) -> str: + """Return the governed study lifecycle status code.""" + return self[3] + + @property + def recorded_from(self) -> datetime: + """Return the exact UTC instant when this version became recorded truth.""" + return self[4] + + @property + def recorded_to(self) -> datetime | None: + """Return the exact UTC close instant when present.""" + return self[5] + + +def _store_view_fields(fields: tuple[tuple[str, object], ...]) -> tuple[tuple[str, object], ...]: + """Store UUID-valued projection fields without retaining mutable UUID object aliases.""" + return tuple( + ( + field_name, + _store_operational_uuid(field_name, value) + if field_name == "criterion_blueprint_id" + else value, + ) + for field_name, value in fields + ) + + +def _restore_view_fields(fields: tuple[tuple[str, object], ...]) -> tuple[tuple[str, object], ...]: + """Return a public projection with fresh UUID objects for UUID-valued fields.""" + return tuple( + ( + field_name, + _restore_operational_uuid(field_name, value) + if field_name == "criterion_blueprint_id" + else value, + ) + for field_name, value in fields + ) + + +class ValidityStudyView(tuple): + """Structurally immutable field-minimized view returned after authorization. + + Tuple-backed storage keeps target UUIDs and UUID-valued projected evidence as + immutable integers, so downstream gateway, audit, or workspace code cannot + rewrite authorized identity through retained UUID objects. The public + constructor is deliberately non-issuing: callers obtain this data-only + projection from ``read_validity_study`` and must re-authorize consequential + actions rather than treating the Python runtime type as a durable credential. + """ + + __slots__ = () + + def __new__( + cls, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + fields: tuple[tuple[str, object], ...], + ) -> ValidityStudyView: + """Reject public construction so only the authorized read path issues views.""" + raise TypeError("ValidityStudyView is issued only by read_validity_study.") + + @property + def tenant_record_id(self) -> UUID: + """Return a fresh tenant identity authorized for this view.""" + return _restore_operational_uuid("tenant_record_id", self[0]) + + @property + def validity_study_id(self) -> UUID: + """Return a fresh validity-study identity authorized for this view.""" + return _restore_operational_uuid("validity_study_id", self[1]) + + @property + def fields(self) -> tuple[tuple[str, object], ...]: + """Return ordered field-minimized evidence with fresh UUID-valued projections.""" + return _restore_view_fields(self[2]) + + +@runtime_checkable +class ValidityStudyReadPort(Protocol): + """Owner repository contract for one tenant-local validity-study header.""" + + def read_validity_study( + self, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + ) -> ValidityStudyRecord | None: + """Return one visible owner record or ``None`` without crossing service tables.""" + ... + + +_PROTOCOL_READ_CAPABILITY = getattr_static(ValidityStudyReadPort, "read_validity_study") + + +def read_validity_study( + *, + principal: ValidationPrincipal, + tenant_record_id: UUID, + validity_study_id: UUID, + purpose_code: str, + requested_fields: frozenset[str], + policy: PurposeBoundAccessPolicy, + read_port: ValidityStudyReadPort, +) -> ValidityStudyView: + """Authorize and read one validity-study header through the canonical owner port. + + Authorization is completed before persistence. The exact ordinary repository + method is captured inertly before authorization and that same function is + invoked after authorization, so dynamic instance lookup cannot switch the + validated capability. Immutable integer snapshots preserve the authorized + target across the executable repository call. The persistence result is + reconstructed into an exact immutable value and must match those snapshots + before any field is returned. + """ + if type(principal) is not ValidationPrincipal: + raise TypeError("principal must be an exact ValidationPrincipal.") + if type(policy) is not PurposeBoundAccessPolicy: + raise TypeError("policy must be an exact PurposeBoundAccessPolicy.") + read_capability = getattr_static(type(read_port), "read_validity_study", None) + if type(read_capability) is not FunctionType or read_capability is _PROTOCOL_READ_CAPABILITY: + raise TypeError("read_port must expose a statically callable read_validity_study.") + + detached_principal = ValidationPrincipal( + tenant_record_id=principal.tenant_record_id, + actor_reference=principal.actor_reference, + granted_scope_codes=principal.granted_scope_codes, + ) + tenant_identity = _store_operational_uuid("tenant_record_id", tenant_record_id) + study_identity = _store_operational_uuid("validity_study_id", validity_study_id) + tenant_id = _restore_operational_uuid("tenant_record_id", tenant_identity) + study_id = _restore_operational_uuid("validity_study_id", study_identity) + purpose = _require_code("purpose_code", purpose_code) + fields = _validate_requested_fields(requested_fields) + detached_policy = _detach_policy(policy) + + require_purpose_bound_access( + request=PurposeBoundAccessRequest( + tenant_record_id=tenant_id, + actor_tenant_record_id=detached_principal.tenant_record_id, + resource_tenant_record_id=tenant_id, + actor_reference=detached_principal.actor_reference, + resource_reference=f"{_RESOURCE_KIND}:{study_id}", + purpose_code=purpose, + operation_code=_OPERATION, + resource_kind=_RESOURCE_KIND, + requested_fields=fields, + granted_scope_codes=detached_principal.granted_scope_codes, + ), + policy=detached_policy, + ) + + persisted = read_capability( + read_port, + tenant_record_id=_restore_operational_uuid("tenant_record_id", tenant_identity), + validity_study_id=_restore_operational_uuid("validity_study_id", study_identity), + ) + if persisted is None: + raise ValidityStudyNotFound(str(study_id)) + if type(persisted) is not ValidityStudyRecord: + raise ValidityStudyIntegrityError("repository returned a non-canonical validity-study record") + + record = ValidityStudyRecord( + tenant_record_id=persisted.tenant_record_id, + validity_study_id=persisted.validity_study_id, + criterion_blueprint_id=persisted.criterion_blueprint_id, + study_status_code=persisted.study_status_code, + recorded_from=persisted.recorded_from, + recorded_to=persisted.recorded_to, + ) + if ( + _store_operational_uuid("record tenant_record_id", record.tenant_record_id) + != tenant_identity + or _store_operational_uuid("record validity_study_id", record.validity_study_id) + != study_identity + ): + raise ValidityStudyIntegrityError("repository returned a validity-study record for another target") + + values = { + "criterion_blueprint_id": record.criterion_blueprint_id, + "study_status_code": record.study_status_code, + "recorded_from": record.recorded_from, + "recorded_to": record.recorded_to, + } + projected_fields = tuple((field_name, values[field_name]) for field_name in sorted(fields)) + return tuple.__new__( + ValidityStudyView, + ( + tenant_identity, + study_identity, + _store_view_fields(projected_fields), + ), + ) diff --git a/services/workforce-validation-api/tests/test_persistence_layout.py b/services/workforce-validation-api/tests/test_persistence_layout.py new file mode 100644 index 00000000..cf77b6bd --- /dev/null +++ b/services/workforce-validation-api/tests/test_persistence_layout.py @@ -0,0 +1,41 @@ +"""Architecture contract for workforce-validation-owned PostgreSQL persistence.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +MIGRATION = ROOT / "services/workforce-validation-api/database/migrations/0001_owner_schema.sql" +FOUNDATION_WORKFLOW = ROOT / ".github/workflows/foundation-ci.yml" +OWNER_SCHEMA_POSTGRES_CONTRACT = "test_workforce_validation_owner_schema_postgres.sh" + + +def test_owner_schema_migration_establishes_deny_default_role_boundary() -> None: + """Require a service-owned schema and least-privilege database role before adapters.""" + sql = MIGRATION.read_text(encoding="utf-8") + + required = ( + "CREATE ROLE workforce_validation_role NOLOGIN", + "CREATE SCHEMA workforce_validation AUTHORIZATION workforce_validation_role", + "REVOKE ALL ON SCHEMA workforce_validation FROM PUBLIC", + ) + for contract in required: + assert contract in sql + + assert "ALTER ROLE workforce_validation_role SET search_path" not in sql + assert "CREATE TABLE" not in sql + assert "public.validity_study" not in sql + assert "GRANT ALL" not in sql + + +def test_owner_migration_history_is_bounded_context_local() -> None: + """Prevent a new global migration number from colliding with other active lanes.""" + relative_path = MIGRATION.relative_to(ROOT).as_posix() + + assert relative_path == "services/workforce-validation-api/database/migrations/0001_owner_schema.sql" + + +def test_owner_schema_postgres_contract_is_admitted_to_foundation() -> None: + """Require the owner-schema bootstrap to execute in the canonical PostgreSQL matrix.""" + workflow = FOUNDATION_WORKFLOW.read_text(encoding="utf-8") + + assert OWNER_SCHEMA_POSTGRES_CONTRACT in workflow diff --git a/services/workforce-validation-api/tests/test_policy_runtime_integrity.py b/services/workforce-validation-api/tests/test_policy_runtime_integrity.py new file mode 100644 index 00000000..c72a6c22 --- /dev/null +++ b/services/workforce-validation-api/tests/test_policy_runtime_integrity.py @@ -0,0 +1,113 @@ +"""Regression for executable policy scalar values at the validation boundary.""" + +from __future__ import annotations + +from uuid import UUID + +import pytest + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_workforce_validation_api.registry import ( + ValidationPrincipal, + ValidityStudyReadPort, + read_validity_study, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +STUDY = UUID("00000000-0000-7000-8000-0000000000c1") + + +class _ExecutableText(str): + """Trip if authorization compares this caller-defined string subtype.""" + + calls = 0 + __hash__ = str.__hash__ + + def __eq__(self, other: object) -> bool: + """Expose any equality comparison before the boundary rejects the subtype.""" + type(self).calls += 1 + raise AssertionError("caller-defined policy comparison executed") + + def __ne__(self, other: object) -> bool: + """Expose any inequality comparison before the boundary rejects the subtype.""" + type(self).calls += 1 + raise AssertionError("caller-defined policy comparison executed") + + +class _ReadPort: + """Record whether persistence was reached.""" + + def __init__(self) -> None: + self.calls = 0 + + def read_validity_study(self, *, tenant_record_id: UUID, validity_study_id: UUID) -> None: + """Fail the test if a rejected policy reaches persistence.""" + del tenant_record_id, validity_study_id + self.calls += 1 + return None + + +def test_policy_text_subtype_is_rejected_before_comparison_or_persistence() -> None: + _ExecutableText.calls = 0 + port = _ReadPort() + assert isinstance(port, ValidityStudyReadPort) + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="validation-read-v1", + resource_kind=_ExecutableText("validity_study_record"), + purpose_code="validation_review", + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset({"study_status_code"}), + ) + + with pytest.raises(ValueError, match="policy resource_kind"): + read_validity_study( + principal=ValidationPrincipal( + tenant_record_id=TENANT, + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=policy, + read_port=port, + ) + + assert _ExecutableText.calls == 0 + assert port.calls == 0 + + +def test_policy_field_subtype_is_rejected_before_comparison_or_persistence() -> None: + _ExecutableText.calls = 0 + port = _ReadPort() + assert isinstance(port, ValidityStudyReadPort) + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="validation-read-v1", + resource_kind="validity_study_record", + purpose_code="validation_review", + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset({_ExecutableText("study_status_code")}), + ) + + with pytest.raises(ValueError, match="policy permitted_fields"): + read_validity_study( + principal=ValidationPrincipal( + tenant_record_id=TENANT, + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=policy, + read_port=port, + ) + + assert _ExecutableText.calls == 0 + assert port.calls == 0 diff --git a/services/workforce-validation-api/tests/test_principal_storage_integrity.py b/services/workforce-validation-api/tests/test_principal_storage_integrity.py new file mode 100644 index 00000000..a0b29985 --- /dev/null +++ b/services/workforce-validation-api/tests/test_principal_storage_integrity.py @@ -0,0 +1,83 @@ +"""Regression contract for canonical validation-principal storage before authorization.""" + +from __future__ import annotations + +from uuid import UUID + +import pytest + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_workforce_validation_api.registry import ( + ValidationPrincipal, + ValidityStudyRecord, + read_validity_study, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +STUDY = UUID("00000000-0000-7000-8000-0000000000c1") + + +class _ExecutableUUID(UUID): + """Expose executable behavior if a UUID subtype reaches downstream validation.""" + + def __getattribute__(self, name: str) -> object: + if name == "int": + raise AttributeError("UUID subtype behavior executed") + return super().__getattribute__(name) + + +class _ReadPort: + """Capture repository use; this regression must fail before persistence.""" + + def __init__(self) -> None: + self.calls: list[tuple[UUID, UUID]] = [] + + def read_validity_study( + self, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + ) -> ValidityStudyRecord | None: + """Record an unexpected persistence call.""" + self.calls.append((tenant_record_id, validity_study_id)) + return None + + +def _policy() -> PurposeBoundAccessPolicy: + """Return the canonical purpose-bound policy used by the read boundary.""" + return PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="validation-read-v1", + resource_kind="validity_study_record", + purpose_code="validation_review", + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset({"study_status_code"}), + ) + + +def test_low_level_exact_principal_is_revalidated_before_keyverse_evaluation() -> None: + """Reject constructor-bypassed identity evidence before subtype behavior can execute.""" + forged_tenant = _ExecutableUUID(str(TENANT)) + principal = tuple.__new__( + ValidationPrincipal, + ( + forged_tenant, + "person:analyst-1", + frozenset({"orgmetra.workforce_validation.read"}), + ), + ) + port = _ReadPort() + + with pytest.raises(ValueError, match="tenant_record_id must be an exact operational UUID"): + read_validity_study( + principal=principal, + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(), + read_port=port, + ) + + assert port.calls == [] diff --git a/services/workforce-validation-api/tests/test_read_port_dependency_integrity.py b/services/workforce-validation-api/tests/test_read_port_dependency_integrity.py new file mode 100644 index 00000000..e49cefc9 --- /dev/null +++ b/services/workforce-validation-api/tests/test_read_port_dependency_integrity.py @@ -0,0 +1,141 @@ +"""Regression contracts for inert repository capability validation before authorization.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_workforce_validation_api.registry import ( + ValidationPrincipal, + ValidityStudyReadPort, + ValidityStudyRecord, + read_validity_study, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +STUDY = UUID("00000000-0000-7000-8000-0000000000c1") +CRITERION = UUID("00000000-0000-7000-8000-0000000000a1") +RECORDED_FROM = datetime(2026, 11, 3, tzinfo=timezone.utc) + + +class _DescriptorReadPort: + """Expose a non-callable static protocol member whose getter must never execute.""" + + @property + def read_validity_study(self) -> object: + """Trip if dependency validation or later code executes this descriptor.""" + raise AssertionError("repository descriptor executed before rejection") + + +class _DynamicLookupReadPort: + """Expose one safe class method but a different callable through instance lookup.""" + + def __init__(self) -> None: + self.dynamic_lookups = 0 + self.static_calls = 0 + + def __getattribute__(self, name: str) -> object: + """Trip if the authorized path performs a second dynamic capability lookup.""" + if name == "read_validity_study": + dynamic_lookups = object.__getattribute__(self, "dynamic_lookups") + object.__setattr__(self, "dynamic_lookups", dynamic_lookups + 1) + + def switched_capability(*, tenant_record_id: UUID, validity_study_id: UUID) -> object: + del tenant_record_id, validity_study_id + raise AssertionError("dynamic repository capability lookup executed after validation") + + return switched_capability + return object.__getattribute__(self, name) + + def read_validity_study( + self, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + ) -> ValidityStudyRecord: + """Return valid owner evidence when the statically validated method is invoked.""" + self.static_calls += 1 + return ValidityStudyRecord( + tenant_record_id=tenant_record_id, + validity_study_id=validity_study_id, + criterion_blueprint_id=CRITERION, + study_status_code="study_draft", + recorded_from=RECORDED_FROM, + recorded_to=None, + ) + + +class _InheritedProtocolReadPort(ValidityStudyReadPort): + """Intentionally inherit the Protocol declaration without implementing persistence.""" + + +def _principal() -> ValidationPrincipal: + """Return one exact authenticated validation principal.""" + return ValidationPrincipal( + tenant_record_id=TENANT, + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ) + + +def _policy(*, purpose_code: str = "validation_review") -> PurposeBoundAccessPolicy: + """Return one purpose-bound policy for the focused repository tests.""" + return PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="validation-read-v1", + resource_kind="validity_study_record", + purpose_code=purpose_code, + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset({"study_status_code"}), + ) + + +def test_noncallable_repository_capability_fails_before_authorization() -> None: + """Reject an invalid port before a deliberately denying policy can be evaluated.""" + with pytest.raises(TypeError, match="read_port must expose a statically callable read_validity_study"): + read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(purpose_code="audit_review"), + read_port=_DescriptorReadPort(), # type: ignore[arg-type] + ) + + +def test_inherited_protocol_placeholder_fails_before_authorization() -> None: + """Require a concrete repository implementation before Keyverse policy evaluation.""" + with pytest.raises(TypeError, match="read_port must expose a statically callable read_validity_study"): + read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(purpose_code="audit_review"), + read_port=_InheritedProtocolReadPort(), + ) + + +def test_validated_repository_capability_is_the_capability_invoked_after_authorization() -> None: + """Bind the inertly validated class method instead of re-resolving it dynamically.""" + port = _DynamicLookupReadPort() + + view = read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(), + read_port=port, + ) + + assert port.dynamic_lookups == 0 + assert port.static_calls == 1 + assert view.fields == (("study_status_code", "study_draft"),) diff --git a/services/workforce-validation-api/tests/test_registry.py b/services/workforce-validation-api/tests/test_registry.py new file mode 100644 index 00000000..53d5a62f --- /dev/null +++ b/services/workforce-validation-api/tests/test_registry.py @@ -0,0 +1,299 @@ +"""Regression contract for the workforce-validation study registry boundary.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from uuid import UUID +from zoneinfo import ZoneInfo + +import pytest + +from orgmetra_keyverse_adapter import AuthorizationDeniedError, PurposeBoundAccessPolicy +from orgmetra_workforce_validation_api.registry import ( + ValidationPrincipal, + ValidityStudyIntegrityError, + ValidityStudyNotFound, + ValidityStudyReadPort, + ValidityStudyRecord, + read_validity_study, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +OTHER_TENANT = UUID("10000000-0000-7000-8000-000000000002") +STUDY = UUID("00000000-0000-7000-8000-0000000000c1") +OTHER_STUDY = UUID("00000000-0000-7000-8000-0000000000c2") +CRITERION = UUID("00000000-0000-7000-8000-0000000000a1") +RECORDED_FROM = datetime(2026, 11, 3, tzinfo=timezone.utc) + + +class _ReadPort: + """Return one configured registry record and capture the authorized target.""" + + def __init__(self, result: object) -> None: + self.result = result + self.calls: list[tuple[UUID, UUID]] = [] + + def read_validity_study(self, *, tenant_record_id: UUID, validity_study_id: UUID) -> object: + """Capture the target and return the configured persistence result.""" + self.calls.append((tenant_record_id, validity_study_id)) + return self.result + + +class _NoReadMethod: + """Deliberately fail the runtime repository protocol.""" + + +def _record(*, tenant_record_id: UUID = TENANT, validity_study_id: UUID = STUDY) -> ValidityStudyRecord: + return ValidityStudyRecord( + tenant_record_id=tenant_record_id, + validity_study_id=validity_study_id, + criterion_blueprint_id=CRITERION, + study_status_code="study_draft", + recorded_from=RECORDED_FROM, + recorded_to=None, + ) + + +def _principal(*, tenant_record_id: UUID = TENANT) -> ValidationPrincipal: + return ValidationPrincipal( + tenant_record_id=tenant_record_id, + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ) + + +def _policy(*, tenant_record_id: UUID = TENANT) -> PurposeBoundAccessPolicy: + return PurposeBoundAccessPolicy( + tenant_record_id=tenant_record_id, + policy_version_code="validation-read-v1", + resource_kind="validity_study_record", + purpose_code="validation_review", + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset( + { + "criterion_blueprint_id", + "study_status_code", + "recorded_from", + "recorded_to", + } + ), + ) + + +def test_read_returns_only_authorized_requested_fields() -> None: + port = _ReadPort(_record()) + + view = read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code", "criterion_blueprint_id"}), + policy=_policy(), + read_port=port, + ) + + assert isinstance(port, ValidityStudyReadPort) + assert port.calls == [(TENANT, STUDY)] + assert view.tenant_record_id == TENANT + assert view.validity_study_id == STUDY + assert view.fields == ( + ("criterion_blueprint_id", CRITERION), + ("study_status_code", "study_draft"), + ) + + +def test_authorization_denial_happens_before_persistence() -> None: + port = _ReadPort(_record()) + + with pytest.raises(AuthorizationDeniedError): + read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"recorded_from"}), + policy=PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="validation-read-v1", + resource_kind="validity_study_record", + purpose_code="audit_review", + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset({"recorded_from"}), + ), + read_port=port, + ) + + assert port.calls == [] + + +def test_missing_study_is_not_found() -> None: + with pytest.raises(ValidityStudyNotFound): + read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(), + read_port=_ReadPort(None), + ) + + +def test_foreign_or_noncanonical_persistence_result_fails_closed() -> None: + for result in (_record(tenant_record_id=OTHER_TENANT), _record(validity_study_id=OTHER_STUDY), object()): + with pytest.raises(ValidityStudyIntegrityError): + read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(), + read_port=_ReadPort(result), + ) + + +def test_dependency_and_request_types_fail_before_repository_use() -> None: + port = _ReadPort(_record()) + common = dict( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(), + read_port=port, + ) + + for key, value, error in ( + ("principal", object(), TypeError), + ("policy", object(), TypeError), + ("read_port", _NoReadMethod(), TypeError), + ("tenant_record_id", "not-a-uuid", ValueError), + ("validity_study_id", UUID(int=0), ValueError), + ("purpose_code", "Validation Review", ValueError), + ("purpose_code", 7, ValueError), + ("requested_fields", set({"study_status_code"}), ValueError), + ("requested_fields", frozenset(), ValueError), + ("requested_fields", frozenset({"unknown_field"}), ValueError), + ("requested_fields", frozenset({7}), ValueError), + ): + arguments = dict(common) + arguments[key] = value + with pytest.raises(error): + read_validity_study(**arguments) + + assert port.calls == [] + + +def test_principal_rejects_invalid_identity_and_scope_shapes() -> None: + invalid_values = ( + dict(tenant_record_id=UUID(int=0), actor_reference="person:analyst-1", granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"})), + dict(tenant_record_id=TENANT, actor_reference="not namespaced", granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"})), + dict(tenant_record_id=TENANT, actor_reference=7, granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"})), + dict(tenant_record_id=TENANT, actor_reference="person:analyst-1", granted_scope_codes=frozenset()), + dict(tenant_record_id=TENANT, actor_reference="person:analyst-1", granted_scope_codes=frozenset({"bad-scope"})), + dict(tenant_record_id=TENANT, actor_reference="person:analyst-1", granted_scope_codes=frozenset({7})), + ) + for values in invalid_values: + with pytest.raises(ValueError): + ValidationPrincipal(**values) + + +def test_principal_is_structurally_immutable_after_identity_validation() -> None: + principal = _principal() + + for field_name, replacement in ( + ("tenant_record_id", OTHER_TENANT), + ("actor_reference", "person:attacker-2"), + ("granted_scope_codes", frozenset({"orgmetra.audit.read"})), + ): + with pytest.raises(AttributeError): + object.__setattr__(principal, field_name, replacement) + + assert principal.tenant_record_id == TENANT + assert principal.actor_reference == "person:analyst-1" + assert principal.granted_scope_codes == frozenset({"orgmetra.workforce_validation.read"}) + + +def test_record_rejects_noncanonical_or_invalid_durable_scalars() -> None: + valid = dict( + tenant_record_id=TENANT, + validity_study_id=STUDY, + criterion_blueprint_id=CRITERION, + study_status_code="study_draft", + recorded_from=RECORDED_FROM, + recorded_to=None, + ) + cases = ( + ("tenant_record_id", UUID(int=0)), + ("validity_study_id", "not-a-uuid"), + ("criterion_blueprint_id", UUID(int=(1 << 128) - 1)), + ("study_status_code", "Study Draft"), + ("study_status_code", 7), + ("recorded_from", datetime(2026, 11, 3)), + ("recorded_to", "not-a-datetime"), + ) + for field_name, value in cases: + arguments = dict(valid) + arguments[field_name] = value + with pytest.raises(ValueError): + ValidityStudyRecord(**arguments) + + with pytest.raises(ValueError): + ValidityStudyRecord(**{**valid, "recorded_to": RECORDED_FROM}) + + +def test_valid_record_detaches_supported_timezones_to_utc() -> None: + for provider in (timezone(timedelta(hours=9)), ZoneInfo("Asia/Seoul")): + record = ValidityStudyRecord( + tenant_record_id=TENANT, + validity_study_id=STUDY, + criterion_blueprint_id=CRITERION, + study_status_code="study_draft", + recorded_from=datetime(2026, 11, 3, 9, tzinfo=provider), + recorded_to=datetime(2026, 11, 4, 9, tzinfo=provider), + ) + + assert type(record.recorded_from) is datetime + assert record.recorded_from.tzinfo is timezone.utc + assert record.recorded_from.hour == 0 + assert record.recorded_to is not None + assert record.recorded_to.tzinfo is timezone.utc + + +def test_record_is_structurally_immutable_against_object_setattr() -> None: + record = _record() + + with pytest.raises(AttributeError): + object.__setattr__(record, "study_status_code", "study_closed") + + assert record.study_status_code == "study_draft" + + +def test_authorized_view_is_structurally_immutable_after_field_minimization() -> None: + view = read_validity_study( + principal=_principal(), + tenant_record_id=TENANT, + validity_study_id=STUDY, + purpose_code="validation_review", + requested_fields=frozenset({"study_status_code"}), + policy=_policy(), + read_port=_ReadPort(_record()), + ) + original_fields = view.fields + + for field_name, replacement in ( + ("tenant_record_id", OTHER_TENANT), + ("validity_study_id", OTHER_STUDY), + ("fields", (("study_status_code", "study_closed"),)), + ): + with pytest.raises(AttributeError): + object.__setattr__(view, field_name, replacement) + + assert view.tenant_record_id == TENANT + assert view.validity_study_id == STUDY + assert view.fields == original_fields diff --git a/services/workforce-validation-api/tests/test_uuid_payload_integrity.py b/services/workforce-validation-api/tests/test_uuid_payload_integrity.py new file mode 100644 index 00000000..5b20e787 --- /dev/null +++ b/services/workforce-validation-api/tests/test_uuid_payload_integrity.py @@ -0,0 +1,30 @@ +"""Regression contract for exact UUID payload validation before sentinel comparison.""" + +from __future__ import annotations + +from uuid import UUID + +import pytest + +from orgmetra_workforce_validation_api.registry import ValidationPrincipal + + +class _ExecutableUUIDPayload: + """Fail if validation compares a forged UUID payload before proving it is an int.""" + + def __eq__(self, other: object) -> bool: + """Expose equality execution as a trust-boundary violation.""" + raise AssertionError(f"forged UUID payload executed equality against {other!r}") + + +def test_exact_uuid_with_executable_internal_payload_fails_before_comparison() -> None: + """Exact UUID outer type cannot authorize executable non-integer internal storage.""" + tenant_record_id = UUID("10000000-0000-7000-8000-000000000001") + object.__setattr__(tenant_record_id, "int", _ExecutableUUIDPayload()) + + with pytest.raises(ValueError, match="tenant_record_id must be an exact operational UUID"): + ValidationPrincipal( + tenant_record_id=tenant_record_id, + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ) diff --git a/services/workforce-validation-api/tests/test_uuid_storage_integrity.py b/services/workforce-validation-api/tests/test_uuid_storage_integrity.py new file mode 100644 index 00000000..52c23998 --- /dev/null +++ b/services/workforce-validation-api/tests/test_uuid_storage_integrity.py @@ -0,0 +1,161 @@ +"""Regression contract for UUID storage behind immutable registry value objects.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_workforce_validation_api.registry import ( + ValidationPrincipal, + ValidityStudyIntegrityError, + ValidityStudyRecord, + read_validity_study, +) + +TENANT_TEXT = "10000000-0000-7000-8000-000000000001" +OTHER_TENANT = UUID("10000000-0000-7000-8000-000000000002") +STUDY_TEXT = "00000000-0000-7000-8000-0000000000c1" +OTHER_STUDY = UUID("00000000-0000-7000-8000-0000000000c2") +CRITERION_TEXT = "00000000-0000-7000-8000-0000000000a1" +OTHER_CRITERION = UUID("00000000-0000-7000-8000-0000000000a2") +RECORDED_FROM = datetime(2026, 11, 3, tzinfo=timezone.utc) + + +class _ReadPort: + """Return one configured owner record for UUID-storage regression coverage.""" + + def __init__(self, result: ValidityStudyRecord) -> None: + self.result = result + + def read_validity_study( + self, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + ) -> ValidityStudyRecord: + """Return the configured record after the application boundary authorizes the read.""" + return self.result + + +class _TargetSwitchingReadPort: + """Attempt to rewrite the authorized UUID target during the executable port call.""" + + def read_validity_study( + self, + *, + tenant_record_id: UUID, + validity_study_id: UUID, + ) -> ValidityStudyRecord: + """Mutate received UUID aliases and return a record matching the rewritten target.""" + object.__setattr__(tenant_record_id, "int", OTHER_TENANT.int) + object.__setattr__(validity_study_id, "int", OTHER_STUDY.int) + return ValidityStudyRecord( + tenant_record_id=OTHER_TENANT, + validity_study_id=OTHER_STUDY, + criterion_blueprint_id=UUID(CRITERION_TEXT), + study_status_code="study_draft", + recorded_from=RECORDED_FROM, + recorded_to=None, + ) + + +def _policy() -> PurposeBoundAccessPolicy: + """Return the canonical purpose-bound policy for the regression read.""" + return PurposeBoundAccessPolicy( + tenant_record_id=UUID(TENANT_TEXT), + policy_version_code="validation-read-v1", + resource_kind="validity_study_record", + purpose_code="validation_review", + operation_code="read", + required_scope_code="orgmetra.workforce_validation.read", + permitted_fields=frozenset({"criterion_blueprint_id"}), + ) + + +def _principal() -> ValidationPrincipal: + """Return one canonical principal for UUID target-integrity tests.""" + return ValidationPrincipal( + tenant_record_id=UUID(TENANT_TEXT), + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ) + + +def test_principal_and_record_do_not_retain_mutable_uuid_inputs() -> None: + """Retained UUID aliases cannot rewrite identity evidence after constructor validation.""" + tenant = UUID(TENANT_TEXT) + study = UUID(STUDY_TEXT) + criterion = UUID(CRITERION_TEXT) + principal = ValidationPrincipal( + tenant_record_id=tenant, + actor_reference="person:analyst-1", + granted_scope_codes=frozenset({"orgmetra.workforce_validation.read"}), + ) + record = ValidityStudyRecord( + tenant_record_id=tenant, + validity_study_id=study, + criterion_blueprint_id=criterion, + study_status_code="study_draft", + recorded_from=RECORDED_FROM, + recorded_to=None, + ) + + object.__setattr__(tenant, "int", OTHER_TENANT.int) + object.__setattr__(study, "int", OTHER_STUDY.int) + object.__setattr__(criterion, "int", OTHER_CRITERION.int) + + assert principal.tenant_record_id == UUID(TENANT_TEXT) + assert record.tenant_record_id == UUID(TENANT_TEXT) + assert record.validity_study_id == UUID(STUDY_TEXT) + assert record.criterion_blueprint_id == UUID(CRITERION_TEXT) + + +def test_port_cannot_switch_the_authorized_target_by_mutating_received_uuid_objects() -> None: + """The target comparison uses pre-port immutable identity evidence, not mutable aliases.""" + with pytest.raises(ValidityStudyIntegrityError, match="another target"): + read_validity_study( + principal=_principal(), + tenant_record_id=UUID(TENANT_TEXT), + validity_study_id=UUID(STUDY_TEXT), + purpose_code="validation_review", + requested_fields=frozenset({"criterion_blueprint_id"}), + policy=_policy(), + read_port=_TargetSwitchingReadPort(), + ) + + +def test_authorized_view_does_not_retain_or_expose_mutable_uuid_storage() -> None: + """Target and projected UUID evidence remain stable across retained-reference rewrites.""" + tenant = UUID(TENANT_TEXT) + study = UUID(STUDY_TEXT) + record = ValidityStudyRecord( + tenant_record_id=UUID(TENANT_TEXT), + validity_study_id=UUID(STUDY_TEXT), + criterion_blueprint_id=UUID(CRITERION_TEXT), + study_status_code="study_draft", + recorded_from=RECORDED_FROM, + recorded_to=None, + ) + + view = read_validity_study( + principal=_principal(), + tenant_record_id=tenant, + validity_study_id=study, + purpose_code="validation_review", + requested_fields=frozenset({"criterion_blueprint_id"}), + policy=_policy(), + read_port=_ReadPort(record), + ) + + object.__setattr__(tenant, "int", OTHER_TENANT.int) + object.__setattr__(study, "int", OTHER_STUDY.int) + projected_criterion = dict(view.fields)["criterion_blueprint_id"] + assert type(projected_criterion) is UUID + object.__setattr__(projected_criterion, "int", OTHER_CRITERION.int) + + assert view.tenant_record_id == UUID(TENANT_TEXT) + assert view.validity_study_id == UUID(STUDY_TEXT) + assert dict(view.fields)["criterion_blueprint_id"] == UUID(CRITERION_TEXT) diff --git a/services/workforce-validation-api/tests/test_view_issuance_integrity.py b/services/workforce-validation-api/tests/test_view_issuance_integrity.py new file mode 100644 index 00000000..a44fce40 --- /dev/null +++ b/services/workforce-validation-api/tests/test_view_issuance_integrity.py @@ -0,0 +1,26 @@ +"""Regression contract for workforce-validation authorized-view issuance.""" + +from uuid import UUID + +import pytest + +import orgmetra_workforce_validation_api.registry as registry + + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +STUDY = UUID("00000000-0000-7000-8000-0000000000c1") + + +def test_direct_authorized_view_construction_fails_closed() -> None: + """Require purpose-bound reads, not public construction, to issue study views.""" + with pytest.raises(TypeError, match="issued only by read_validity_study"): + registry.ValidityStudyView( + tenant_record_id=TENANT, + validity_study_id=STUDY, + fields=(("study_status_code", "study_draft"),), + ) + + +def test_registry_module_exposes_no_unconditional_view_issuer() -> None: + """Keep ordinary view issuance inside the authorized read application path.""" + assert not hasattr(registry, "_issue_validity_study_view") diff --git a/tests/test_foundation_ci_dependency_hygiene.sh b/tests/test_foundation_ci_dependency_hygiene.sh index 2c0f5087..d214695f 100644 --- a/tests/test_foundation_ci_dependency_hygiene.sh +++ b/tests/test_foundation_ci_dependency_hygiene.sh @@ -18,6 +18,7 @@ expected_pythonpaths=( "packages/selection-review/src" "services/job-analysis-api/src:packages/hris-kernel/src:packages/keyverse-adapter/src" "services/people-api/src:packages/hris-kernel/src:packages/keyverse-adapter/src" + "services/workforce-validation-api/src:packages/keyverse-adapter/src" ) if ! grep -Fq -- "${expected_install}" "${workflow_path}"; then diff --git a/tests/test_workforce_validation_owner_schema_postgres.sh b/tests/test_workforce_validation_owner_schema_postgres.sh new file mode 100644 index 00000000..c79b659c --- /dev/null +++ b/tests/test_workforce_validation_owner_schema_postgres.sh @@ -0,0 +1,76 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${DATABASE_URL:=postgresql://orgmetra:orgmetra@localhost:5432/orgmetra}" + +migration="services/workforce-validation-api/database/migrations/0001_owner_schema.sql" +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f "${migration}" + +role_flags="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SELECT rolcanlogin, rolsuper, rolcreatedb, rolcreaterole, rolinherit, rolreplication, rolbypassrls +FROM pg_roles +WHERE rolname = 'workforce_validation_role'; +")" +if [[ "${role_flags}" != "f|f|f|f|f|f|f" ]]; then + echo "workforce_validation_role flags are not deny-default: ${role_flags}" >&2 + exit 1 +fi + +schema_owner="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SELECT pg_get_userbyid(nspowner) +FROM pg_namespace +WHERE nspname = 'workforce_validation'; +")" +if [[ "${schema_owner}" != "workforce_validation_role" ]]; then + echo "workforce_validation schema has unexpected owner: ${schema_owner}" >&2 + exit 1 +fi + +role_config="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SELECT COALESCE(array_to_string(rolconfig, ','), '') +FROM pg_roles +WHERE rolname = 'workforce_validation_role'; +")" +if [[ -n "${role_config}" ]]; then + echo "NOLOGIN schema owner must not carry ineffective login-only runtime defaults: ${role_config}" >&2 + exit 1 +fi + +set_role_probe="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SET search_path = public; +SET ROLE workforce_validation_role; +SELECT current_user || '|' || current_setting('search_path'); +RESET ROLE; +")" +if [[ "${set_role_probe}" != "workforce_validation_role|public" ]]; then + echo "unexpected SET ROLE search_path behavior: ${set_role_probe}" >&2 + exit 1 +fi + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -qc "CREATE ROLE workforce_validation_public_probe NOLOGIN;" +trap 'psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -qc "DROP ROLE IF EXISTS workforce_validation_public_probe;" >/dev/null 2>&1 || true' EXIT + +public_usage="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SELECT has_schema_privilege('workforce_validation_public_probe', 'workforce_validation', 'USAGE'); +")" +public_create="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SELECT has_schema_privilege('workforce_validation_public_probe', 'workforce_validation', 'CREATE'); +")" +if [[ "${public_usage}" != "f" || "${public_create}" != "f" ]]; then + echo "PUBLIC retains workforce_validation schema privileges: usage=${public_usage} create=${public_create}" >&2 + exit 1 +fi + +relation_count="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -Atqc " +SELECT count(*) +FROM pg_class AS relation +JOIN pg_namespace AS namespace ON namespace.oid = relation.relnamespace +WHERE namespace.nspname = 'workforce_validation'; +")" +if [[ "${relation_count}" != "0" ]]; then + echo "owner-schema bootstrap created application relations prematurely: ${relation_count}" >&2 + exit 1 +fi + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -qc "DROP ROLE workforce_validation_public_probe;" +trap - EXIT diff --git a/tests/validate_repository.py b/tests/validate_repository.py index d9d4c15a..1a05c8ef 100644 --- a/tests/validate_repository.py +++ b/tests/validate_repository.py @@ -88,6 +88,7 @@ "tests/test_audit_outbox_hardening_postgres.sh", "tests/test_candidate_worker_conversion_postgres.sh", "tests/test_validity_study_case_postgres.sh", + "tests/test_workforce_validation_owner_schema_postgres.sh", "tests/test_criterion_observation_scope_postgres.sh", "tests/test_people_mutation_idempotency_postgres.sh", "tests/test_job_analysis_snapshot_postgres.sh", @@ -634,4 +635,4 @@ def main() -> None: if __name__ == "__main__": - main() + main() \ No newline at end of file