diff --git a/docs/adr/0149-employee-profile-employment-history-read.md b/docs/adr/0149-employee-profile-employment-history-read.md new file mode 100644 index 000000000..7049bd435 --- /dev/null +++ b/docs/adr/0149-employee-profile-employment-history-read.md @@ -0,0 +1,34 @@ +# ADR 0149: Purpose-bound employee Employment-history read + +- **Status:** Accepted on active PR #149; not protected-main truth until integrated. +- **Date:** 2026-08-29 +- **Owners:** Orgmetra People API / HRIS core +- **Extends:** ADR 0003 (bitemporal HRIS data), ADR 0008 (purpose-bound PII authorization) + +## Decision + +The employee profile reads Employment history through a read-only People API service boundary that authorizes the exact tenant, Person, purpose, operation, and requested field set **before** calling the injected persistence port. The persistence adapter remains a separate port; this slice does not create a second Employment source of truth and does not introduce cross-service application-table SQL. + +Each persistence row carries a durable Employment identity, a durable Employment-version identity, controlled Employment status and concurrency codes, business-effective (`effective_from`, `effective_to`) coordinates, and system-recorded (`recorded_from`, `recorded_to`) coordinates. `known_at` selects the half-open recorded interval `[recorded_from, recorded_to)`. Trust-bearing system instants require an exact built-in `datetime` using Python's built-in fixed-offset `timezone` at zero offset so validation and canonical rendering cannot depend on mutable caller-defined timezone behavior. + +Persistence output is untrusted. Exact tuple and row types are only shape checks, so the accepted `EmploymentHistoryRecord` itself uses tuple-backed immutable storage: a persistence adapter retaining the returned row cannot rewrite its fields in place through ordinary assignment or `object.__setattr__`. The service still reconstructs every accepted persistence row through the public validating constructor before tenant/Person scope, system-time visibility, version uniqueness, business-time overlap checks, deterministic sorting, or authorized field emission. That reconstruction remains necessary because low-level tuple construction can bypass the public validating constructor; forged or malformed exact-type rows therefore fail closed at runtime integrity validation rather than becoming authorized output. + +Structural in-process immutability is not a substitute for a transactional database snapshot, MVCC, row/version locking, or the persistence adapter's own consistency guarantees. A future database adapter remains responsible for returning one transactionally coherent bitemporal view at the requested knowledge cutoff. + +Only policy-authorized fields are emitted. Employment identity and version identity are not unconditional response-envelope fields; a caller authorized only for status receives status only. This preserves field minimization and prevents identifiers from becoming a side channel around purpose-bound authorization. + +## Security and privacy consequences + +Authorization is resource-centric and per request. NIST SP 800-207 and SP 800-207A support resource/service authorization decisions independent of network location; Orgmetra applies that principle through the existing Keyverse adapter contract rather than embedding another policy engine. + +The persistence-alias boundary is treated as a local integrity concern rather than as evidence that the persistence adapter is malicious. Structurally immutable row storage closes the in-process validation-to-use alias rewrite path at the object boundary, while service-owned reconstruction preserves fail-closed validation even for deliberately forged low-level tuple instances. This does not add cross-service locks or weaken field minimization. + +The read boundary does not infer attendance, availability, fitness, compensation, performance, candidate status, or employment-decision authority. It exposes only authoritative Employment facts already permitted by policy for the requested Person and system-time cutoff. + +## Data consequences + +This ADR preserves the normalized distinction among Person, Employment, Organization, Job, Position, and Assignment. It changes no database schema. A future PostgreSQL adapter must remain tenant-scoped and RLS-governed and must read only Orgmetra-owned Employment tables through the People service boundary. + +## Verification + +PR #149 must demonstrate authorization-before-retrieval, tenant/Person isolation, half-open system-time visibility, controlled codes, exact UUID/time validation, field minimization, schema/type drift failure, structural resistance to retained-alias rewriting, fail-closed revalidation of low-level forged exact-type rows, duplicate-version rejection, business-time overlap rejection, deterministic ordering, exact 100% owned People API statement/branch coverage, and all applicable repository/security/central gates before integration. diff --git a/docs/doctoring/employee-profile-employment-history-references.md b/docs/doctoring/employee-profile-employment-history-references.md new file mode 100644 index 000000000..759704c3a --- /dev/null +++ b/docs/doctoring/employee-profile-employment-history-references.md @@ -0,0 +1,21 @@ +# Employee profile Employment-history read — primary references + +## Current authoritative references + +National Institute of Standards and Technology. (2020). *Zero trust architecture* (NIST Special Publication 800-207). U.S. Department of Commerce. https://doi.org/10.6028/NIST.SP.800-207 + +National Institute of Standards and Technology. (2023). *A zero trust architecture model for access control in cloud-native applications in multi-cloud environments* (NIST Special Publication 800-207A). U.S. Department of Commerce. https://doi.org/10.6028/NIST.SP.800-207A + +PostgreSQL Global Development Group. (2025). *PostgreSQL 18 documentation: Date/time types*. https://www.postgresql.org/docs/18/datatype-datetime.html + +## Why these sources matter to PR #149 + +NIST SP 800-207 treats access to enterprise resources as an explicit authentication/authorization decision rather than an implicit consequence of network location. SP 800-207A extends granular identity-based enforcement to application and service boundaries. PR #149 applies that principle narrowly by authorizing the exact tenant, Person, purpose, operation, and requested Employment-history fields before protected retrieval. + +PostgreSQL 18's date/time semantics support the repository's existing separation of business dates from timezone-aware system-recorded instants. The PR does not claim that PostgreSQL prescribes Orgmetra's bitemporal domain model; ADR 0003 remains the product architecture authority. + +These references support the authorization and temporal representation boundaries only. They do not establish NIST certification, PostgreSQL conformance certification, or authority to infer attendance, fitness, compensation, performance, or an employment decision from Employment history. + +## Review date + +Rechecked as current final primary references on 2026-08-29. Re-review if NIST publishes a superseding final zero-trust application authorization specification or the protected repository changes its supported PostgreSQL major version. diff --git a/docs/traceability/employee-profile-employment-history-read.md b/docs/traceability/employee-profile-employment-history-read.md new file mode 100644 index 000000000..301ee4dae --- /dev/null +++ b/docs/traceability/employee-profile-employment-history-read.md @@ -0,0 +1,49 @@ +# Employee profile Employment-history read traceability + +## Product requirement + +Protected Orgmetra planning requires an authoritative bitemporal HRIS core and buyer-readable employee history. Protected `develop` already stores `employment_record` and `employment_record_version` truth and exposes governed People reads, but before PR #149 it has no purpose-bound service contract for returning Employment history at an explicit system knowledge cutoff. + +## Protected-main truth consumed + +- `database/migrations/0001_foundation_schema.sql` separates `employment_record` identity from bitemporal `employment_record_version` business/system truth. +- `services/people-api/src/orgmetra_people_api/authorization.py` delegates protected-field authorization to the integrated purpose-bound Keyverse adapter contract. +- `services/people-api/src/orgmetra_people_api/people.py` establishes authorization-before-protected-read and target-scope revalidation. +- `services/people-api/src/orgmetra_people_api/mutations.py` defines current controlled Employment statuses (`active`, `leave`, `terminated`) and concurrency codes (`exclusive`, `concurrent`). +- `.github/workflows/people-api-quality.yml` requires exact 100% owned People API statement and branch coverage. + +## PR #149 active implementation + +| Requirement | Production boundary | Regression evidence | +| --- | --- | --- | +| Authorize before protected retrieval | `read_employment_history()` calls `authorize_resource_fields()` before `EmploymentHistoryReadPort` | denied-field test requires zero port calls | +| Preserve business/system time separately | `EmploymentHistoryRecord.effective_*` and `.recorded_*` | deterministic history and recorded-cutoff tests | +| Tenant/Person isolation | service revalidates every service-owned reconstruction | other-tenant and other-Person rows fail closed | +| Half-open system visibility | `[recorded_from, recorded_to)` at exact `known_at` | future-recorded and `recorded_to == known_at` rows fail closed | +| Controlled Employment semantics | exact built-in status/concurrency codes | unknown and string-subclass codes fail closed | +| Field minimization | output built only from `decision.authorized_fields` | status-only grant leaks no Employment identity | +| No reflective schema expansion | explicit supported-field encoder requires exact built-in `str` | unknown and string-subclass fields fail closed | +| Persistence runtime integrity | exact tuple + exact row type + validating service-owned reconstruction | mutable container, unsupported row, and low-level forged exact-type regressions | +| Validation-to-use alias integrity | `EmploymentHistoryRecord` uses tuple-backed immutable storage and `_snapshot_persistence_record()` reconstructs before use | `object.__setattr__` rewrite attempts fail at the row boundary; forged low-level tuple instances fail runtime integrity | +| Version integrity | unique `employment_record_version_id` per response snapshot | duplicate version identity fails closed | +| Bitemporal business integrity | visible snapshots of one Employment cannot overlap effective time | overlapping intervals fail closed; adjacent intervals remain valid | +| Deterministic history | sort by effective start, Employment UUID, version UUID | reversed persistence order returns canonical order | +| Trust-bearing identity/time integrity | exact operational UUIDs and built-in UTC instants | sentinel/subclass UUID and malformed system time fail before protected retrieval or row use | + +## Scope exclusions + +PR #149 does not create/update/delete Employment, alter schema, expose a PostgreSQL adapter, add UI geometry, infer attendance/fitness/compensation/performance, or authorize an employment decision. It does not mutate Keyverse or any other dedicated-writer repository. A future persistence adapter and employee-profile UI must reuse this contract instead of bypassing the People service. + +The in-process row object is structurally immutable, and the service revalidates a detached reconstruction before authorization output. This does not claim to replace database transaction isolation, MVCC, locks, or a persistence adapter's obligation to return one coherent view. + +## Test-first evidence rule + +Contract head `23c3417edd7024ecc4c1c64f2d7017b573ab9eaf` added the original executable regression before production `employment_history.py` existed. Hosted execution for that predecessor was queued when the implementation branch advanced, so queued/cancelled predecessor evidence is **not** represented as a terminal RED. The contract-first source ordering remains auditable in Git history. + +A later integrity review identified a second, narrower validation-to-use defect: the service revalidated the exact persistence-owned `EmploymentHistoryRecord` and then retained that same object for overlap checks and authorized encoding. Because `object.__setattr__` can rewrite a frozen dataclass through an alias, a holder of the persistence row could change an already-validated value before use. Exact head `5cdbeb2028a49bd0277159a03042c5d95dd2a06d` added the realistic alias-rewrite regression before the root repair; the production repair begins at `45b4ff5ec9fb065a665e1fe51bc2120d46cdc62a` by reconstructing a service-owned validated snapshot and discarding the persistence alias for subsequent decisions and output. + +A third integrity review identified a capture-window defect in that repair: one sequential reconstruction could read an old value for one field and a concurrently rewritten value for a later field, producing a valid-looking service-owned row that never existed as one source state. Exact head `6eb105d6310adbdb9e33f64fab4cd450a9681968` added `test_alias_rewrite_during_snapshot_cannot_create_torn_authorized_row` before the production change. Its workflows were still queued when the branch advanced, so no terminal RED is claimed. The prior repair beginning at `4dfbd2a9f32947e5c1c61d6eccee47b57781dc92` required two consecutive validated captures to compare equal. + +A fourth integrity review identified the remaining root weakness: the double-capture guard still accepted a record type whose storage itself could be rewritten through `object.__setattr__`, leaving correctness dependent on detecting mutation after the fact. Exact test-only head `c07ce7baf738679e1ef5cbef1d98760fefe670e3` added `test_persistence_record_is_structurally_immutable_against_object_setattr`. People API Quality run `33257244737`, exact checkout job `99113016031`, produced a genuine terminal RED: 1 failed / 159 passed, with the new test failing because `object.__setattr__` did **not** raise; owned production coverage remained 1524/1524 statements and 508/508 branches. The root repair begins at `6ef636cdf803ef3195f80db089f1ee432e0d7646`: `EmploymentHistoryRecord` moves to tuple-backed immutable storage, while service-owned reconstruction continues to validate low-level exact-type instances that bypass the public constructor. + +Only tests/checks bound to the final unchanged PR #149 head are passing integration evidence. Queued, pending, skipped, cancelled, absent, predecessor-head, status-only, or model-only evidence is non-passing, and another PR's checks/reviews never transfer. diff --git a/services/people-api/src/orgmetra_people_api/__init__.py b/services/people-api/src/orgmetra_people_api/__init__.py index b043bed33..8486d5552 100644 --- a/services/people-api/src/orgmetra_people_api/__init__.py +++ b/services/people-api/src/orgmetra_people_api/__init__.py @@ -7,6 +7,14 @@ extract_bearer_token, ) from orgmetra_people_api.authorization import authorize_resource_fields +from orgmetra_people_api.employment_history import ( + AuthorizedEmploymentHistoryEntry, + AuthorizedEmploymentHistoryView, + EmploymentHistoryIntegrityError, + EmploymentHistoryReadPort, + EmploymentHistoryRecord, + read_employment_history, +) from orgmetra_people_api.hire import ( HireAcceptanceCommand, HireAcceptancePort, @@ -45,13 +53,22 @@ from orgmetra_people_api.postgres_mutations import PostgresPeopleMutationPort __all__ = [ + "AssignmentMutationCommand", + "AssignmentMutationResult", "AuthenticatedPrincipal", "AuthenticationFailed", + "AuthorizedEmploymentHistoryEntry", + "AuthorizedEmploymentHistoryView", "AuthorizedWorkerPeopleView", + "EmploymentHistoryIntegrityError", + "EmploymentHistoryReadPort", + "EmploymentHistoryRecord", + "EmploymentMutationCommand", + "EmploymentMutationResult", + "HireAcceptanceAsgiApp", "HireAcceptanceCommand", "HireAcceptancePort", "HireAcceptanceResult", - "HireAcceptanceAsgiApp", "HireDecisionIntegrityError", "HireDecisionNotFound", "PeopleAsgiApp", @@ -67,10 +84,6 @@ "PostgresHireAcceptancePort", "PostgresPeopleMutationPort", "PostgresPeopleReadPort", - "AssignmentMutationCommand", - "AssignmentMutationResult", - "EmploymentMutationCommand", - "EmploymentMutationResult", "TokenAuthenticator", "WorkerPeopleRecord", "accept_confirmed_hire", @@ -79,5 +92,6 @@ "create_employment_record", "create_position_record", "extract_bearer_token", + "read_employment_history", "read_worker_people_record", ] diff --git a/services/people-api/src/orgmetra_people_api/employment_history.py b/services/people-api/src/orgmetra_people_api/employment_history.py new file mode 100644 index 000000000..3350310a4 --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/employment_history.py @@ -0,0 +1,324 @@ +"""Purpose-bound bitemporal Employment-history reads for the employee profile. + +Authorization is completed before an injected persistence port may retrieve +protected Employment facts. Persistence output is treated as untrusted and is +revalidated for tenant/person scope, business-time consistency, and system-time +visibility before any authorized values are returned. +""" + +from __future__ import annotations + +from collections import namedtuple +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from typing import Protocol, runtime_checkable +from uuid import UUID + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy + +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.authorization import authorize_resource_fields + +_MAX_UUID_INT = (1 << 128) - 1 +_EMPLOYMENT_STATUSES = frozenset({"active", "leave", "terminated"}) +_CONCURRENCY_CODES = frozenset({"exclusive", "concurrent"}) + + +class EmploymentHistoryIntegrityError(RuntimeError): + """Indicate that persistence violated the authorized Employment-history contract.""" + + +def _validate_operational_uuid(field_name: str, value: object) -> None: + """Require an exact UUID outside Orgmetra's reserved protocol sentinels.""" + if type(value) is not UUID or value.int in (0, _MAX_UUID_INT): + raise ValueError(f"{field_name} must be an operational UUID.") + + +def _validate_utc_instant(field_name: str, value: object) -> None: + """Require an exact datetime with Python's deterministic built-in UTC timezone.""" + if ( + type(value) is not datetime + or type(value.tzinfo) is not timezone + or value.utcoffset() != timedelta(0) + ): + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + + +_EmploymentHistoryRecordTuple = namedtuple( + "_EmploymentHistoryRecordTuple", + ( + "tenant_record_id", + "person_record_id", + "employment_record_id", + "employment_record_version_id", + "employment_status_code", + "employment_concurrency_code", + "effective_from", + "effective_to", + "recorded_from", + "recorded_to", + ), +) + + +class EmploymentHistoryRecord(_EmploymentHistoryRecordTuple): + """One structurally immutable Employment version at a system-time cutoff.""" + + __slots__ = () + + tenant_record_id: UUID + person_record_id: UUID + employment_record_id: UUID + employment_record_version_id: UUID + employment_status_code: str + employment_concurrency_code: str + effective_from: date + effective_to: date | None + recorded_from: datetime + recorded_to: datetime | None + + def __new__( + cls, + *, + tenant_record_id: UUID, + person_record_id: UUID, + employment_record_id: UUID, + employment_record_version_id: UUID, + employment_status_code: str, + employment_concurrency_code: str, + effective_from: date, + effective_to: date | None, + recorded_from: datetime, + recorded_to: datetime | None, + ) -> EmploymentHistoryRecord: + """Build one validated row whose tuple storage cannot be rewritten in place.""" + instance = super().__new__( + cls, + tenant_record_id, + person_record_id, + employment_record_id, + employment_record_version_id, + employment_status_code, + employment_concurrency_code, + effective_from, + effective_to, + recorded_from, + recorded_to, + ) + instance.assert_runtime_integrity() + return instance + + def assert_runtime_integrity(self) -> None: + """Revalidate a row after it crosses the untrusted persistence boundary.""" + for field_name in ( + "tenant_record_id", + "person_record_id", + "employment_record_id", + "employment_record_version_id", + ): + _validate_operational_uuid(field_name, getattr(self, field_name)) + if type(self.employment_status_code) is not str or self.employment_status_code not in _EMPLOYMENT_STATUSES: + raise ValueError("employment_status_code must be active, leave, or terminated.") + if ( + type(self.employment_concurrency_code) is not str + or self.employment_concurrency_code not in _CONCURRENCY_CODES + ): + raise ValueError("employment_concurrency_code must be exclusive or concurrent.") + if type(self.effective_from) is not date: + raise ValueError("effective_from must be a business date.") + if self.effective_to is not None and ( + type(self.effective_to) is not date or self.effective_to <= self.effective_from + ): + raise ValueError("effective_to must be later than effective_from when present.") + _validate_utc_instant("recorded_from", self.recorded_from) + if self.recorded_to is not None: + _validate_utc_instant("recorded_to", self.recorded_to) + if self.recorded_to <= self.recorded_from: + raise ValueError("recorded_to must be later than recorded_from when present.") + + +@runtime_checkable +class EmploymentHistoryReadPort(Protocol): + """Read tenant/person-scoped Employment versions at one system knowledge cutoff.""" + + def read_employment_history( + self, + *, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + ) -> tuple[EmploymentHistoryRecord, ...]: + """Return Employment rows visible to persistence at ``known_at``.""" + + +@dataclass(frozen=True, slots=True) +class AuthorizedEmploymentHistoryEntry: + """One Employment version containing only explicitly authorized fields.""" + + field_values: tuple[tuple[str, str | None], ...] + + +@dataclass(frozen=True, slots=True) +class AuthorizedEmploymentHistoryView: + """Purpose-bound employee-profile Employment-history response.""" + + resource_reference: str + entries: tuple[AuthorizedEmploymentHistoryEntry, ...] + + +def _instant_text(value: datetime) -> str: + """Render a validated UTC instant in one canonical RFC 3339 representation.""" + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _authorized_field_value(record: EmploymentHistoryRecord, field_name: str) -> str | None: + """Return one explicitly supported Employment-history field without reflection.""" + if type(field_name) is not str: + raise EmploymentHistoryIntegrityError("authorization returned an unsupported Employment-history field") + if field_name == "effective_from": + return record.effective_from.isoformat() + if field_name == "effective_to": + return None if record.effective_to is None else record.effective_to.isoformat() + if field_name == "employment_concurrency_code": + return record.employment_concurrency_code + if field_name == "employment_record_id": + return str(record.employment_record_id) + if field_name == "employment_record_version_id": + return str(record.employment_record_version_id) + if field_name == "employment_status_code": + return record.employment_status_code + if field_name == "recorded_from": + return _instant_text(record.recorded_from) + if field_name == "recorded_to": + return None if record.recorded_to is None else _instant_text(record.recorded_to) + raise EmploymentHistoryIntegrityError("authorization returned an unsupported Employment-history field") + + +def _is_recorded_visible(record: EmploymentHistoryRecord, known_at: datetime) -> bool: + """Return whether ``known_at`` lies in the row's half-open system interval.""" + return record.recorded_from <= known_at and (record.recorded_to is None or known_at < record.recorded_to) + + +def _reject_effective_overlap(records: list[EmploymentHistoryRecord]) -> None: + """Reject overlapping business-time truth for one Employment at one knowledge cutoff.""" + previous_by_employment: dict[UUID, EmploymentHistoryRecord] = {} + for record in sorted(records, key=lambda item: (item.employment_record_id.int, item.effective_from)): + previous = previous_by_employment.get(record.employment_record_id) + if previous is not None and ( + previous.effective_to is None or record.effective_from < previous.effective_to + ): + raise EmploymentHistoryIntegrityError("overlapping Employment business-time truth") + previous_by_employment[record.employment_record_id] = record + + +def _capture_persistence_record(record: EmploymentHistoryRecord) -> EmploymentHistoryRecord: + """Reconstruct and validate one persistence-owned Employment row.""" + return EmploymentHistoryRecord( + tenant_record_id=record.tenant_record_id, + person_record_id=record.person_record_id, + employment_record_id=record.employment_record_id, + employment_record_version_id=record.employment_record_version_id, + employment_status_code=record.employment_status_code, + employment_concurrency_code=record.employment_concurrency_code, + effective_from=record.effective_from, + effective_to=record.effective_to, + recorded_from=record.recorded_from, + recorded_to=record.recorded_to, + ) + + +def _snapshot_persistence_record(record: EmploymentHistoryRecord) -> EmploymentHistoryRecord: + """Detach and revalidate structurally immutable persistence evidence. + + ``EmploymentHistoryRecord`` stores its fields in immutable tuple storage, so a + persistence adapter retaining the returned object cannot rewrite that alias in + place through ``object.__setattr__``. Reconstruction is still mandatory because + low-level tuple construction can bypass the public validating constructor. + + This in-process integrity boundary does not replace a transactional database + snapshot, MVCC, locking, or the persistence layer's own concurrency controls. + """ + return _capture_persistence_record(record) + + +def read_employment_history( + *, + principal: AuthenticatedPrincipal, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + purpose_code: str, + requested_fields: frozenset[str], + policy: PurposeBoundAccessPolicy, + read_port: EmploymentHistoryReadPort, +) -> AuthorizedEmploymentHistoryView: + """Authorize then return bitemporal Employment history for one Person. + + A denied purpose, scope, target, or field request causes zero protected reads. + After retrieval, every row is detached from its persistence-owned alias and + must still match the authorized tenant/person and requested system-time view. + Duplicate version identities and overlapping business-time truth for one + Employment fail closed instead of being guessed. + """ + _validate_operational_uuid("tenant_record_id", tenant_record_id) + _validate_operational_uuid("person_record_id", person_record_id) + _validate_utc_instant("known_at", known_at) + + resource_reference = f"person_employment_history:{person_record_id.hex}" + decision = authorize_resource_fields( + principal=principal, + tenant_record_id=tenant_record_id, + resource_tenant_record_id=tenant_record_id, + resource_reference=resource_reference, + purpose_code=purpose_code, + operation_code="read_record", + resource_kind="person_employment_history", + requested_fields=requested_fields, + policy=policy, + ) + + records = read_port.read_employment_history( + tenant_record_id=tenant_record_id, + person_record_id=person_record_id, + known_at=known_at, + ) + if type(records) is not tuple: + raise EmploymentHistoryIntegrityError("Employment-history persistence must return an immutable tuple") + + seen_version_ids: set[UUID] = set() + verified: list[EmploymentHistoryRecord] = [] + for record in records: + if type(record) is not EmploymentHistoryRecord: + raise EmploymentHistoryIntegrityError("Employment-history persistence returned an unsupported row type") + try: + trusted_record = _snapshot_persistence_record(record) + except ValueError as exc: + raise EmploymentHistoryIntegrityError("Employment-history row failed runtime integrity") from exc + if trusted_record.tenant_record_id != tenant_record_id or trusted_record.person_record_id != person_record_id: + raise EmploymentHistoryIntegrityError("Employment-history row does not match the authorized target") + if not _is_recorded_visible(trusted_record, known_at): + raise EmploymentHistoryIntegrityError("Employment-history row is not visible at the requested knowledge cutoff") + if trusted_record.employment_record_version_id in seen_version_ids: + raise EmploymentHistoryIntegrityError("duplicate Employment version identity") + seen_version_ids.add(trusted_record.employment_record_version_id) + verified.append(trusted_record) + + _reject_effective_overlap(verified) + authorized_fields = tuple(sorted(decision.authorized_fields)) + entries = tuple( + AuthorizedEmploymentHistoryEntry( + field_values=tuple( + (field_name, _authorized_field_value(record, field_name)) + for field_name in authorized_fields + ) + ) + for record in sorted( + verified, + key=lambda item: ( + item.effective_from, + item.employment_record_id.int, + item.employment_record_version_id.int, + ), + ) + ) + return AuthorizedEmploymentHistoryView(resource_reference=decision.resource_reference, entries=entries) diff --git a/services/people-api/tests/test_employment_history_alias_integrity.py b/services/people-api/tests/test_employment_history_alias_integrity.py new file mode 100644 index 000000000..296606e44 --- /dev/null +++ b/services/people-api/tests/test_employment_history_alias_integrity.py @@ -0,0 +1,153 @@ +"""Regressions for persistence-held Employment-history row aliases. + +A persistence adapter may retain a reference to a row after returning it. The +People boundary therefore uses structurally immutable tuple storage for accepted +rows and reconstructs each row through the validating constructor before use. +Low-level tuple construction remains untrusted and must fail closed when invalid. +""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from unittest.mock import patch +from uuid import UUID + +import pytest + +import orgmetra_people_api.employment_history as employment_history_module +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal + +TENANT = UUID("0198a412-7100-7000-8000-000000000001") +PERSON = UUID("0198a412-7100-7000-8000-000000000010") +EMPLOYMENT = UUID("0198a412-7100-7000-8000-000000000020") +VERSION = UUID("0198a412-7100-7000-8000-000000000030") +KNOWN_AT = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc) + + +class AliasHoldingPort: + """Return a row while deliberately retaining the persistence-side alias.""" + + def __init__(self, record: employment_history_module.EmploymentHistoryRecord) -> None: + self.record = record + + def read_employment_history( + self, + *, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + ) -> tuple[employment_history_module.EmploymentHistoryRecord, ...]: + """Return the retained row exactly as an in-process adapter could.""" + return (self.record,) + + +def _principal() -> AuthenticatedPrincipal: + """Return an authorized HR operator for Employment-history regressions.""" + return AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse:hr-operator", + granted_scope_codes=frozenset({"orgmetra.people.employment_history.read"}), + ) + + +def _policy(*fields: str) -> PurposeBoundAccessPolicy: + """Return one purpose-bound Employment-history read policy.""" + return PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="employee-profile-employment-history-v1", + resource_kind="person_employment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.employment_history.read", + permitted_fields=frozenset(fields), + ) + + +def _record() -> employment_history_module.EmploymentHistoryRecord: + """Return one valid immutable Employment-history record.""" + return employment_history_module.EmploymentHistoryRecord( + tenant_record_id=TENANT, + person_record_id=PERSON, + employment_record_id=EMPLOYMENT, + employment_record_version_id=VERSION, + employment_status_code="active", + employment_concurrency_code="exclusive", + effective_from=date(2025, 1, 1), + effective_to=None, + recorded_from=datetime(2026, 8, 20, 0, 0, tzinfo=timezone.utc), + recorded_to=None, + ) + + +def test_persistence_record_is_structurally_immutable_against_object_setattr() -> None: + """A retained persistence alias must not support low-level in-place field rewriting.""" + record = _record() + + with pytest.raises((AttributeError, TypeError)): + object.__setattr__(record, "employment_status_code", "leave") + + assert record.employment_status_code == "active" + + +def test_persistence_alias_cannot_rewrite_authorized_value_after_validation() -> None: + """A post-validation alias rewrite attempt must fail before response serialization.""" + record = _record() + policy = _policy("employment_status_code") + original_overlap_check = employment_history_module._reject_effective_overlap + + def reject_retained_alias_rewrite( + records: list[employment_history_module.EmploymentHistoryRecord], + ) -> None: + """Attempt the retained-alias attack after all row validation has completed.""" + original_overlap_check(records) + with pytest.raises((AttributeError, TypeError)): + object.__setattr__(record, "employment_status_code", "leave") + + with patch.object( + employment_history_module, + "_reject_effective_overlap", + side_effect=reject_retained_alias_rewrite, + ): + view = employment_history_module.read_employment_history( + principal=_principal(), + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=policy.permitted_fields, + policy=policy, + read_port=AliasHoldingPort(record), + ) + + assert record.employment_status_code == "active" + assert view.entries[0].field_values == (("employment_status_code", "active"),) + + +def test_low_level_invalid_tuple_reconstruction_fails_runtime_integrity() -> None: + """Bypassing the public constructor must not make forged persistence evidence trusted.""" + record = _record() + raw_values = list(record) + raw_values[4] = "forged" + forged = tuple.__new__( + employment_history_module.EmploymentHistoryRecord, + tuple(raw_values), + ) + policy = _policy("employment_status_code", "employment_concurrency_code") + + assert type(forged) is employment_history_module.EmploymentHistoryRecord + assert forged.employment_status_code == "forged" + with pytest.raises( + employment_history_module.EmploymentHistoryIntegrityError, + match="runtime integrity", + ): + employment_history_module.read_employment_history( + principal=_principal(), + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=policy.permitted_fields, + policy=policy, + read_port=AliasHoldingPort(forged), + ) diff --git a/services/people-api/tests/test_employment_history_read.py b/services/people-api/tests/test_employment_history_read.py new file mode 100644 index 000000000..9b7e1a26d --- /dev/null +++ b/services/people-api/tests/test_employment_history_read.py @@ -0,0 +1,371 @@ +"""Executable contracts for purpose-bound bitemporal Employment history reads.""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import AuthorizationDeniedError, PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.employment_history import ( + EmploymentHistoryIntegrityError, + EmploymentHistoryRecord, + read_employment_history, +) + +TENANT = UUID("0198a412-7100-7000-8000-000000000001") +OTHER_TENANT = UUID("0198a412-7100-7000-8000-000000000002") +PERSON = UUID("0198a412-7100-7000-8000-000000000010") +OTHER_PERSON = UUID("0198a412-7100-7000-8000-000000000011") +EMPLOYMENT_A = UUID("0198a412-7100-7000-8000-000000000020") +EMPLOYMENT_B = UUID("0198a412-7100-7000-8000-000000000021") +VERSION_A = UUID("0198a412-7100-7000-8000-000000000030") +VERSION_B = UUID("0198a412-7100-7000-8000-000000000031") +KNOWN_AT = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc) +RECORDED_FROM = datetime(2026, 8, 20, 0, 0, tzinfo=timezone.utc) + + +class ForgedUUID(UUID): + """Prove trust-bearing identities reject UUID subclasses.""" + + +class ForgedField(str): + """Prove authorization output cannot smuggle behavior in string subclasses.""" + + +class ForgedStatus(str): + """Prove controlled Employment codes require exact built-in strings.""" + + +class FakeEmploymentHistoryPort: + """Capture reads so tests prove authorization occurs before protected retrieval.""" + + def __init__(self, records: object) -> None: + self.records = records + self.calls: list[tuple[UUID, UUID, datetime]] = [] + + def read_employment_history( + self, + *, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + ) -> object: + """Return configured persistence output after recording the exact scope.""" + self.calls.append((tenant_record_id, person_record_id, known_at)) + return self.records + + +def employment_record( + *, + employment_record_id: UUID = EMPLOYMENT_A, + employment_record_version_id: UUID = VERSION_A, + tenant_record_id: UUID = TENANT, + person_record_id: UUID = PERSON, + employment_status_code: str = "active", + employment_concurrency_code: str = "exclusive", + effective_from: date = date(2025, 1, 1), + effective_to: date | None = None, + recorded_from: datetime = RECORDED_FROM, + recorded_to: datetime | None = None, +) -> EmploymentHistoryRecord: + """Build one persisted Employment version for service-contract tests.""" + return EmploymentHistoryRecord( + tenant_record_id=tenant_record_id, + person_record_id=person_record_id, + employment_record_id=employment_record_id, + employment_record_version_id=employment_record_version_id, + employment_status_code=employment_status_code, + employment_concurrency_code=employment_concurrency_code, + effective_from=effective_from, + effective_to=effective_to, + recorded_from=recorded_from, + recorded_to=recorded_to, + ) + + +class EmploymentHistoryReadTests(unittest.TestCase): + """Prove Employment history remains purpose-bound, bitemporal, and deterministic.""" + + def setUp(self) -> None: + self.principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse:hr-operator", + granted_scope_codes=frozenset({"orgmetra.people.employment_history.read"}), + ) + self.policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="employee-profile-employment-history-v1", + resource_kind="person_employment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.employment_history.read", + permitted_fields=frozenset( + { + "employment_record_id", + "employment_record_version_id", + "employment_status_code", + "employment_concurrency_code", + "effective_from", + "effective_to", + "recorded_from", + "recorded_to", + } + ), + ) + + def test_returns_authorized_history_in_deterministic_effective_order(self) -> None: + later = employment_record( + employment_record_id=EMPLOYMENT_B, + employment_record_version_id=VERSION_B, + employment_status_code="leave", + employment_concurrency_code="concurrent", + effective_from=date(2026, 7, 1), + ) + earlier = employment_record( + employment_status_code="terminated", + effective_from=date(2025, 1, 1), + effective_to=date(2026, 6, 30), + recorded_to=datetime(2026, 8, 30, 0, 0, tzinfo=timezone.utc), + ) + port = FakeEmploymentHistoryPort((later, earlier)) + + view = read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=self.policy.permitted_fields, + policy=self.policy, + read_port=port, + ) + + self.assertEqual(view.resource_reference, f"person_employment_history:{PERSON.hex}") + self.assertEqual(port.calls, [(TENANT, PERSON, KNOWN_AT)]) + rows = tuple(dict(entry.field_values) for entry in view.entries) + self.assertEqual(tuple(row["employment_record_id"] for row in rows), (str(EMPLOYMENT_A), str(EMPLOYMENT_B))) + self.assertEqual(rows[0]["employment_record_version_id"], str(VERSION_A)) + self.assertEqual(rows[0]["employment_status_code"], "terminated") + self.assertEqual(rows[0]["employment_concurrency_code"], "exclusive") + self.assertEqual(rows[0]["effective_to"], "2026-06-30") + self.assertEqual(rows[0]["recorded_to"], "2026-08-30T00:00:00Z") + self.assertEqual(rows[1]["employment_status_code"], "leave") + self.assertIsNone(rows[1]["effective_to"]) + self.assertIsNone(rows[1]["recorded_to"]) + + def test_field_minimization_never_leaks_employment_identity(self) -> None: + limited_policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="minimal-v1", + resource_kind="person_employment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.employment_history.read", + permitted_fields=frozenset({"employment_status_code"}), + ) + view = read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"employment_status_code"}), + policy=limited_policy, + read_port=FakeEmploymentHistoryPort((employment_record(),)), + ) + self.assertEqual(view.entries[0].field_values, (("employment_status_code", "active"),)) + + def test_denied_field_never_reaches_employment_repository(self) -> None: + port = FakeEmploymentHistoryPort((employment_record(),)) + limited_policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="limited-v1", + resource_kind="person_employment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.employment_history.read", + permitted_fields=frozenset({"effective_from"}), + ) + with self.assertRaises(AuthorizationDeniedError): + read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"employment_status_code"}), + policy=limited_policy, + read_port=port, + ) + self.assertEqual(port.calls, []) + + def test_repository_scope_or_recorded_visibility_mismatch_fails_closed(self) -> None: + cases = ( + employment_record(tenant_record_id=OTHER_TENANT), + employment_record(person_record_id=OTHER_PERSON), + employment_record(recorded_from=datetime(2026, 8, 30, 0, 0, tzinfo=timezone.utc)), + employment_record(recorded_to=KNOWN_AT), + ) + for record in cases: + with self.subTest(record=record), self.assertRaises(EmploymentHistoryIntegrityError): + read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"effective_from"}), + policy=self.policy, + read_port=FakeEmploymentHistoryPort((record,)), + ) + + def test_repository_container_or_row_type_drift_fails_closed(self) -> None: + for records, message in ( + ([employment_record()], "immutable tuple"), + ((object(),), "unsupported row type"), + ): + with self.subTest(records=records), self.assertRaisesRegex(EmploymentHistoryIntegrityError, message): + read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"effective_from"}), + policy=self.policy, + read_port=FakeEmploymentHistoryPort(records), + ) + + def test_low_level_invalid_row_reconstruction_fails_runtime_integrity(self) -> None: + record = employment_record() + raw_values = list(record) + raw_values[4] = "forged" + forged = tuple.__new__(EmploymentHistoryRecord, tuple(raw_values)) + self.assertIs(type(forged), EmploymentHistoryRecord) + with self.assertRaisesRegex(EmploymentHistoryIntegrityError, "runtime integrity"): + read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"employment_status_code"}), + policy=self.policy, + read_port=FakeEmploymentHistoryPort((forged,)), + ) + + def test_duplicate_version_identity_or_overlapping_effective_truth_fails_closed(self) -> None: + duplicate_version = employment_record(effective_from=date(2026, 7, 1)) + overlap = employment_record( + employment_record_version_id=VERSION_B, + effective_from=date(2025, 6, 1), + effective_to=date(2025, 12, 1), + ) + base = employment_record(effective_to=date(2026, 1, 1)) + for records, message in ( + ((base, duplicate_version), "duplicate Employment version identity"), + ((base, overlap), "overlapping Employment business-time truth"), + ): + with self.subTest(message=message), self.assertRaisesRegex(EmploymentHistoryIntegrityError, message): + read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"effective_from"}), + policy=self.policy, + read_port=FakeEmploymentHistoryPort(records), + ) + + def test_adjacent_versions_for_one_employment_are_valid_history(self) -> None: + first = employment_record(effective_to=date(2026, 1, 1)) + second = employment_record( + employment_record_version_id=VERSION_B, + employment_status_code="leave", + effective_from=date(2026, 1, 1), + ) + view = read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"employment_status_code", "effective_from"}), + policy=self.policy, + read_port=FakeEmploymentHistoryPort((second, first)), + ) + self.assertEqual(tuple(dict(item.field_values)["employment_status_code"] for item in view.entries), ("active", "leave")) + + def test_policy_schema_drift_or_forged_field_fails_closed(self) -> None: + for field in ("future_sensitive_field", ForgedField("effective_from")): + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="drifted-v1", + resource_kind="person_employment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.employment_history.read", + permitted_fields=frozenset({field}), + ) + with self.subTest(field_type=type(field).__name__), self.assertRaisesRegex( + EmploymentHistoryIntegrityError, "unsupported Employment-history field" + ): + read_employment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({field}), + policy=policy, + read_port=FakeEmploymentHistoryPort((employment_record(),)), + ) + + def test_invalid_request_shape_fails_before_repository_access(self) -> None: + port = FakeEmploymentHistoryPort(()) + invalid = ( + ("tenant_record_id", UUID(int=0)), + ("tenant_record_id", ForgedUUID(str(TENANT))), + ("person_record_id", UUID(int=(1 << 128) - 1)), + ("known_at", datetime(2026, 8, 29, 0, 0)), + ) + for field_name, value in invalid: + kwargs = { + "principal": self.principal, + "tenant_record_id": TENANT, + "person_record_id": PERSON, + "known_at": KNOWN_AT, + "purpose_code": "employee_profile_review", + "requested_fields": frozenset({"effective_from"}), + "policy": self.policy, + "read_port": port, + } + kwargs[field_name] = value + with self.subTest(field_name=field_name), self.assertRaises(ValueError): + read_employment_history(**kwargs) + self.assertEqual(port.calls, []) + + def test_record_rejects_noncanonical_identity_code_business_or_system_time(self) -> None: + invalid_overrides = ( + {"employment_record_id": "employment"}, + {"employment_record_version_id": ForgedUUID(str(VERSION_A))}, + {"employment_status_code": "unknown"}, + {"employment_status_code": ForgedStatus("active")}, + {"employment_concurrency_code": "shared"}, + {"employment_concurrency_code": ForgedStatus("exclusive")}, + {"effective_from": datetime(2025, 1, 1, tzinfo=timezone.utc)}, + {"effective_to": date(2025, 1, 1)}, + {"recorded_from": datetime(2026, 8, 20, 0, 0)}, + {"recorded_to": datetime(2026, 8, 21, 0, 0)}, + {"recorded_to": RECORDED_FROM}, + ) + for override in invalid_overrides: + with self.subTest(override=override), self.assertRaises(ValueError): + employment_record(**override) + + +if __name__ == "__main__": + unittest.main()