diff --git a/docs/adr/0142-employee-profile-assignment-history-read.md b/docs/adr/0142-employee-profile-assignment-history-read.md new file mode 100644 index 000000000..47123dce6 --- /dev/null +++ b/docs/adr/0142-employee-profile-assignment-history-read.md @@ -0,0 +1,39 @@ +# ADR 0142: Purpose-bound employee assignment-history read + +- **Status:** Accepted on active PR #142; not protected-main truth until integrated. +- **Date:** 2026-08-28 +- **Owners:** Orgmetra People API / HRIS core +- **Extends:** ADR 0003 (bitemporal HRIS data), ADR 0008 (purpose-bound PII authorization) + +## Decision + +The employee profile reads assignment history through a read-only People API boundary that authorizes the exact tenant, person, purpose, operation, and requested field set **before** calling the persistence adapter. The persistence adapter remains injected; this slice does not introduce direct SQL or a second Assignment source of truth. + +Each returned row carries separate business-effective (`effective_from`, `effective_to`) 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` paired with Python's built-in fixed-offset `timezone` at zero offset; caller-defined `tzinfo` providers are rejected before protected retrieval or row use so validation, comparison, and canonical rendering cannot depend on mutable user-supplied timezone behavior. Orgmetra then revalidates tenant/person scope, recorded-time visibility, row type, and assignment identity uniqueness because persistence output is untrusted at the service boundary. + +Only fields granted by the purpose-bound authorization decision are emitted. Assignment identity is **not** an unconditional envelope field: a caller that is authorized only for `effective_from` receives only `effective_from`. This prevents a row identifier from becoming an accidental side channel around field minimization. + +Results are deterministically ordered by business-effective start and assignment UUID. Allocation values use an exact four-decimal `Decimal` representation in `(0, 1.0000]`; the read path does not infer FTE semantics beyond the authoritative Assignment fact. + +## Security and privacy consequences + +This boundary follows resource-centric, per-request authorization. NIST SP 800-207 describes authorization before establishing access to an enterprise resource, while SP 800-207A extends identity-based granular policy enforcement to application and service boundaries. The implementation therefore delegates policy evaluation to Orgmetra's existing Keyverse adapter contract rather than embedding a second authorization engine. + +The response intentionally excludes display name, contact data, compensation, ratings, assessments, candidate data, credentials, prompts, and model output. A caller may request only explicitly supported assignment-history fields, and schema drift fails closed. + +## Data consequences + +This ADR does not change the normalized Job / Position / Employment / Assignment model. It exposes historical Assignment versions for one authorized person while preserving the distinction between business time and system-recorded time required by ADR 0003. A future PostgreSQL adapter must remain tenant-scoped/RLS-governed and must not bypass the People service with cross-service application-table SQL. + +## Verification + +PR #142 must demonstrate: + +1. denied fields cause zero persistence calls; +2. tenant/person or system-time mismatches fail closed; +3. caller-defined UTC-looking timezone providers fail closed before protected retrieval or persisted row use; +4. duplicate visible assignment identities fail closed; +5. only authorized fields are returned; +6. canonical allocation/time representations and deterministic ordering; +7. exact 100% owned People API statement and branch coverage on the current PR head; +8. applicable Foundation, SAST, Security, Recovery, and central required-workflow evidence before any integration decision. diff --git a/docs/doctoring/employee-profile-assignment-history-references.md b/docs/doctoring/employee-profile-assignment-history-references.md new file mode 100644 index 000000000..d445daf1c --- /dev/null +++ b/docs/doctoring/employee-profile-assignment-history-references.md @@ -0,0 +1,17 @@ +# Employee profile assignment-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 + +## Why these sources matter to PR #142 + +NIST SP 800-207 treats authentication and authorization as resource-access decisions rather than implicit consequences of network location. SP 800-207A extends granular identity-based policy enforcement to application/service boundaries. PR #142 applies that principle narrowly: the People API authorizes the exact tenant, person, purpose, operation, and requested fields before the protected assignment-history port is called. + +These references support the **authorization boundary**, not a claim of NIST certification or compliance. Orgmetra's bitemporal representation remains governed by its own accepted ADR 0003 and authoritative Assignment model. The references do not justify deriving employment decisions, ratings, compensation actions, or inferred worker characteristics from history records. + +## Review date + +Rechecked against official NIST publication pages on 2026-08-28. Re-review these references if NIST publishes a superseding final revision that materially changes application-level authorization guidance. diff --git a/docs/traceability/employee-profile-assignment-history-read.md b/docs/traceability/employee-profile-assignment-history-read.md new file mode 100644 index 000000000..7f0a7cdd1 --- /dev/null +++ b/docs/traceability/employee-profile-assignment-history-read.md @@ -0,0 +1,36 @@ +# Employee profile assignment-history read traceability + +## Product requirement + +Protected `docs/PRD.md` lists **Employee profile with bitemporal assignment history** as a P1 HRIS requirement. PR #142 owns only the backend governed-read slice needed to expose Assignment history to an employee-profile surface. + +## Protected-main truth consumed + +- `packages/hris-kernel/src/orgmetra_hris_kernel/facts.py` defines immutable `AssignmentFact` identity, Employment/Person/Position binding, exact allocation, business-effective interval, and system-recorded interval. +- `services/people-api/src/orgmetra_people_api/authorization.py` delegates field access to the existing purpose-bound Keyverse adapter contract. +- `services/people-api/src/orgmetra_people_api/people.py` establishes the People API pattern that authorizes before protected repository access and revalidates resolved target scope. +- `.github/workflows/people-api-quality.yml` requires exact 100% People API statement and branch coverage. + +## PR #142 active implementation + +| Requirement | Production boundary | Regression evidence | +| --- | --- | --- | +| Authorize before protected retrieval | `read_assignment_history()` calls `authorize_resource_fields()` before `AssignmentHistoryReadPort` | denied-field test asserts zero port calls | +| Preserve business and system time separately | `AssignmentHistoryRecord.effective_*` and `.recorded_*` | visible/history ordering and recorded-cutoff regressions | +| Deterministic UTC trust boundary | `known_at` and recorded instants require an exact `datetime` using Python's built-in fixed-offset `timezone` at zero offset | caller-defined UTC-looking `tzinfo` providers fail before protected retrieval and at persistence-row construction | +| Tenant/person isolation | service revalidates every returned row | other-tenant and other-person rows fail closed | +| Half-open system-time visibility | `[recorded_from, recorded_to)` at exact `known_at` | future-recorded and `recorded_to == known_at` rows fail closed | +| Field minimization | output is built only from `decision.authorized_fields` | effective-only policy does not leak assignment identity | +| No reflective schema expansion | explicit supported-field encoder requires an exact built-in `str` | unknown fields and string-subclass fields fail closed | +| Deterministic history | sort by effective start then assignment UUID | reversed persistence order produces deterministic business-time order | +| Exact allocation evidence | finite four-decimal `Decimal` in `(0, 1.0000]` | NaN, zero, >1, and noncanonical scale rejected | +| Trust-bearing identity integrity | operational identity requires exact built-in `UUID`, not subclasses or sentinel values | UUID subclasses plus Nil/Max sentinels are rejected | +| Persistence runtime integrity | exact tuple + exact row type + `AssignmentHistoryRecord.assert_runtime_integrity()` immediately after retrieval | mutable container, unsupported row type, and post-construction NaN reinjection fail closed | + +## Scope exclusions + +PR #142 does **not** create or change Assignment, Employment, Position, Person, candidate, compensation, performance, or decision records. It adds no UI geometry and does not write to Keyverse or another CWL repository. A PostgreSQL persistence adapter and employee-profile UI wiring remain separate follow-on work and must reuse this contract rather than bypass it. + +## Merge evidence rule + +Only evidence bound to the final unchanged PR #142 head is applicable. Pending, queued, skipped, cancelled, predecessor-head, status-only, or model-only evidence is non-passing. Reviews/checks from another PR do not 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..ca2ebf252 100644 --- a/services/people-api/src/orgmetra_people_api/__init__.py +++ b/services/people-api/src/orgmetra_people_api/__init__.py @@ -1,5 +1,13 @@ """Request-edge, governed read, confirmed-hire, and People mutation contracts.""" +from orgmetra_people_api.assignment_history import ( + AssignmentHistoryIntegrityError, + AssignmentHistoryReadPort, + AssignmentHistoryRecord, + AuthorizedAssignmentHistoryEntry, + AuthorizedAssignmentHistoryView, + read_assignment_history, +) from orgmetra_people_api.auth import ( AuthenticatedPrincipal, AuthenticationFailed, @@ -45,13 +53,22 @@ from orgmetra_people_api.postgres_mutations import PostgresPeopleMutationPort __all__ = [ + "AssignmentHistoryIntegrityError", + "AssignmentHistoryReadPort", + "AssignmentHistoryRecord", + "AssignmentMutationCommand", + "AssignmentMutationResult", "AuthenticatedPrincipal", "AuthenticationFailed", + "AuthorizedAssignmentHistoryEntry", + "AuthorizedAssignmentHistoryView", "AuthorizedWorkerPeopleView", + "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_assignment_history", "read_worker_people_record", ] diff --git a/services/people-api/src/orgmetra_people_api/assignment_history.py b/services/people-api/src/orgmetra_people_api/assignment_history.py new file mode 100644 index 000000000..ede426279 --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/assignment_history.py @@ -0,0 +1,243 @@ +"""Purpose-bound bitemporal assignment-history reads for the employee profile. + +This module exposes a read-only service boundary. Authorization is completed +before the injected persistence port may retrieve protected assignment facts. +The service then verifies tenant/person scope and system-time visibility before +returning only fields granted by the purpose-bound policy decision. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from decimal import Decimal +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 +_ZERO = Decimal("0.0000") +_ONE = Decimal("1.0000") + + +class AssignmentHistoryIntegrityError(RuntimeError): + """Indicate that assignment-history persistence violated the authorized read contract.""" + + +def _validate_operational_uuid(field_name: str, value: object) -> None: + """Require an exact UUID that is not an Orgmetra protocol sentinel.""" + 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 a built-in deterministic zero-offset 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.") + + +@dataclass(frozen=True, slots=True) +class AssignmentHistoryRecord: + """One persistence result for an assignment version visible at a knowledge cutoff. + + ``effective_*`` describes business time. ``recorded_*`` describes the + half-open system-recorded interval during which this version is known. + Allocation is deliberately canonicalized to four decimal places so the + response cannot produce multiple textual identities for one FTE value. + """ + + tenant_record_id: UUID + assignment_record_id: UUID + employment_record_id: UUID + person_record_id: UUID + position_record_id: UUID + allocation_ratio: Decimal + effective_from: date + effective_to: date | None + recorded_from: datetime + recorded_to: datetime | None + + def __post_init__(self) -> None: + """Reject malformed identity, temporal, and allocation evidence at construction.""" + self.assert_runtime_integrity() + + def assert_runtime_integrity(self) -> None: + """Revalidate trust-bearing fields after persistence crosses the service boundary. + + ``frozen=True`` prevents ordinary assignment but is not a security boundary: + hostile or buggy same-process code can still mutate an instance through + low-level Python mechanisms. The read service therefore repeats these + checks immediately before using persistence evidence. + """ + for field_name in ( + "tenant_record_id", + "assignment_record_id", + "employment_record_id", + "person_record_id", + "position_record_id", + ): + _validate_operational_uuid(field_name, getattr(self, field_name)) + if ( + type(self.allocation_ratio) is not Decimal + or not self.allocation_ratio.is_finite() + or self.allocation_ratio <= _ZERO + or self.allocation_ratio > _ONE + or self.allocation_ratio.as_tuple().exponent != -4 + ): + raise ValueError("allocation_ratio must be a finite Decimal in (0, 1.0000] with four decimal places.") + 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 AssignmentHistoryReadPort(Protocol): + """Read tenant/person-scoped assignment versions at one system knowledge cutoff.""" + + def read_assignment_history( + self, + *, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + ) -> tuple[AssignmentHistoryRecord, ...]: + """Return assignment rows visible to the persistence adapter at ``known_at``.""" + + +@dataclass(frozen=True, slots=True) +class AuthorizedAssignmentHistoryEntry: + """One assignment row containing only fields authorized for the stated purpose.""" + + field_values: tuple[tuple[str, str | None], ...] + + +@dataclass(frozen=True, slots=True) +class AuthorizedAssignmentHistoryView: + """Purpose-bound employee-profile assignment history response.""" + + resource_reference: str + entries: tuple[AuthorizedAssignmentHistoryEntry, ...] + + +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: AssignmentHistoryRecord, field_name: str) -> str | None: + """Return one explicitly supported assignment-history field without reflection.""" + if type(field_name) is not str: + raise AssignmentHistoryIntegrityError("authorization returned an unsupported assignment-history field") + if field_name == "allocation_ratio": + return format(record.allocation_ratio, "f") + if field_name == "assignment_record_id": + return str(record.assignment_record_id) + 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_record_id": + return str(record.employment_record_id) + if field_name == "position_record_id": + return str(record.position_record_id) + 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 AssignmentHistoryIntegrityError("authorization returned an unsupported assignment-history field") + + +def _is_recorded_visible(record: AssignmentHistoryRecord, known_at: datetime) -> bool: + """Return whether ``known_at`` lies in the record's half-open system interval.""" + return record.recorded_from <= known_at and (record.recorded_to is None or known_at < record.recorded_to) + + +def read_assignment_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: AssignmentHistoryReadPort, +) -> AuthorizedAssignmentHistoryView: + """Authorize then return the worker's bitemporal assignment history. + + The service never retrieves protected rows when purpose, scope, resource, or + requested fields are denied. Persistence output is treated as untrusted: a + row from another tenant/person, outside the requested recorded-time view, a + post-construction-invalid row, or a duplicate visible assignment identity + fails closed before any values are returned to the caller. + """ + _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_assignment_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_assignment_history", + requested_fields=requested_fields, + policy=policy, + ) + + records = read_port.read_assignment_history( + tenant_record_id=tenant_record_id, + person_record_id=person_record_id, + known_at=known_at, + ) + if type(records) is not tuple: + raise AssignmentHistoryIntegrityError("assignment-history persistence must return an immutable tuple") + + seen_assignment_ids: set[UUID] = set() + verified: list[AssignmentHistoryRecord] = [] + for record in records: + if type(record) is not AssignmentHistoryRecord: + raise AssignmentHistoryIntegrityError("assignment-history persistence returned an unsupported row type") + try: + record.assert_runtime_integrity() + except ValueError as exc: + raise AssignmentHistoryIntegrityError("assignment-history row failed runtime integrity") from exc + if record.tenant_record_id != tenant_record_id or record.person_record_id != person_record_id: + raise AssignmentHistoryIntegrityError("assignment-history row does not match the authorized target") + if not _is_recorded_visible(record, known_at): + raise AssignmentHistoryIntegrityError("assignment-history row is not visible at the requested knowledge cutoff") + if record.assignment_record_id in seen_assignment_ids: + raise AssignmentHistoryIntegrityError("duplicate visible assignment identity") + seen_assignment_ids.add(record.assignment_record_id) + verified.append(record) + + authorized_fields = tuple(sorted(decision.authorized_fields)) + entries = tuple( + AuthorizedAssignmentHistoryEntry( + 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.assignment_record_id.int)) + ) + return AuthorizedAssignmentHistoryView(resource_reference=decision.resource_reference, entries=entries) diff --git a/services/people-api/tests/test_assignment_history_read.py b/services/people-api/tests/test_assignment_history_read.py new file mode 100644 index 000000000..21a2dea98 --- /dev/null +++ b/services/people-api/tests/test_assignment_history_read.py @@ -0,0 +1,358 @@ +"""Executable contracts for purpose-bound bitemporal assignment history reads.""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from decimal import Decimal +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import AuthorizationDeniedError, PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.assignment_history import ( + AssignmentHistoryIntegrityError, + AssignmentHistoryRecord, + read_assignment_history, +) + +TENANT = UUID("0198a412-7000-7000-8000-000000000001") +OTHER_TENANT = UUID("0198a412-7000-7000-8000-000000000002") +PERSON = UUID("0198a412-7000-7000-8000-000000000010") +OTHER_PERSON = UUID("0198a412-7000-7000-8000-000000000011") +EMPLOYMENT = UUID("0198a412-7000-7000-8000-000000000020") +POSITION_A = UUID("0198a412-7000-7000-8000-000000000030") +POSITION_B = UUID("0198a412-7000-7000-8000-000000000031") +ASSIGNMENT_A = UUID("0198a412-7000-7000-8000-000000000040") +ASSIGNMENT_B = UUID("0198a412-7000-7000-8000-000000000041") +KNOWN_AT = datetime(2026, 8, 28, 3, 0, tzinfo=timezone.utc) +RECORDED_FROM = datetime(2026, 8, 20, 0, 0, tzinfo=timezone.utc) + + +class ForgedUUID(UUID): + """Prove trust-bearing identity validators reject subclass behavior.""" + + +class ForgedField(str): + """Prove authorization output cannot smuggle behavior in a string subclass.""" + + +class FakeAssignmentHistoryPort: + """Capture read calls so tests can prove authorization-before-retrieval ordering.""" + + def __init__(self, records: object) -> None: + self.records = records + self.calls: list[tuple[UUID, UUID, datetime]] = [] + + def read_assignment_history( + self, + *, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + ) -> object: + """Return configured persistence output after recording the exact read scope.""" + self.calls.append((tenant_record_id, person_record_id, known_at)) + return self.records + + +def assignment_record( + *, + assignment_record_id: UUID = ASSIGNMENT_A, + tenant_record_id: UUID = TENANT, + person_record_id: UUID = PERSON, + position_record_id: UUID = POSITION_A, + effective_from: date = date(2026, 1, 1), + effective_to: date | None = date(2026, 7, 1), + recorded_from: datetime = RECORDED_FROM, + recorded_to: datetime | None = None, + allocation_ratio: Decimal = Decimal("1.0000"), +) -> AssignmentHistoryRecord: + """Build one persisted assignment-history row for service-contract tests.""" + return AssignmentHistoryRecord( + tenant_record_id=tenant_record_id, + assignment_record_id=assignment_record_id, + employment_record_id=EMPLOYMENT, + person_record_id=person_record_id, + position_record_id=position_record_id, + allocation_ratio=allocation_ratio, + effective_from=effective_from, + effective_to=effective_to, + recorded_from=recorded_from, + recorded_to=recorded_to, + ) + + +class AssignmentHistoryReadTests(unittest.TestCase): + """Prove employee-profile history remains authorized, 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.assignment_history.read"}), + ) + self.policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="employee-profile-assignment-history-v1", + resource_kind="person_assignment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.assignment_history.read", + permitted_fields=frozenset( + { + "assignment_record_id", + "employment_record_id", + "position_record_id", + "allocation_ratio", + "effective_from", + "effective_to", + "recorded_from", + "recorded_to", + } + ), + ) + + def test_returns_authorized_history_in_deterministic_effective_order(self) -> None: + later = assignment_record( + assignment_record_id=ASSIGNMENT_B, + position_record_id=POSITION_B, + effective_from=date(2026, 7, 1), + effective_to=None, + allocation_ratio=Decimal("0.5000"), + ) + earlier = assignment_record(recorded_to=KNOWN_AT.replace(day=30)) + port = FakeAssignmentHistoryPort((later, earlier)) + + view = read_assignment_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_assignment_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["assignment_record_id"] for row in rows), (str(ASSIGNMENT_A), str(ASSIGNMENT_B))) + self.assertEqual(rows[0]["effective_from"], "2026-01-01") + self.assertEqual(rows[0]["effective_to"], "2026-07-01") + self.assertEqual(rows[0]["allocation_ratio"], "1.0000") + self.assertEqual(rows[0]["recorded_to"], "2026-08-30T03:00:00Z") + self.assertIsNone(rows[1]["effective_to"]) + self.assertIsNone(rows[1]["recorded_to"]) + + def test_field_minimization_never_leaks_assignment_identity(self) -> None: + port = FakeAssignmentHistoryPort((assignment_record(),)) + limited_policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="minimal-v1", + resource_kind="person_assignment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.assignment_history.read", + permitted_fields=frozenset({"effective_from"}), + ) + + view = read_assignment_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=limited_policy, + read_port=port, + ) + + self.assertEqual(view.entries[0].field_values, (("effective_from", "2026-01-01"),)) + + def test_denied_field_never_reaches_assignment_repository(self) -> None: + port = FakeAssignmentHistoryPort((assignment_record(),)) + limited_policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="limited-v1", + resource_kind="person_assignment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.assignment_history.read", + permitted_fields=frozenset({"effective_from"}), + ) + + with self.assertRaises(AuthorizationDeniedError): + read_assignment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"position_record_id"}), + policy=limited_policy, + read_port=port, + ) + + self.assertEqual(port.calls, []) + + def test_repository_scope_or_recorded_visibility_mismatch_fails_closed(self) -> None: + future_recorded = KNOWN_AT.replace(day=29) + cases = ( + assignment_record(tenant_record_id=OTHER_TENANT), + assignment_record(person_record_id=OTHER_PERSON), + assignment_record(recorded_from=future_recorded), + assignment_record(recorded_to=KNOWN_AT), + ) + for record in cases: + with self.subTest(record=record): + with self.assertRaises(AssignmentHistoryIntegrityError): + read_assignment_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=FakeAssignmentHistoryPort((record,)), + ) + + def test_repository_container_or_row_type_drift_fails_closed(self) -> None: + for records, message in ( + ([assignment_record()], "immutable tuple"), + ((object(),), "unsupported row type"), + ): + with self.subTest(records=records), self.assertRaisesRegex(AssignmentHistoryIntegrityError, message): + read_assignment_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=FakeAssignmentHistoryPort(records), + ) + + def test_post_construction_row_mutation_fails_runtime_integrity(self) -> None: + record = assignment_record() + object.__setattr__(record, "allocation_ratio", Decimal("NaN")) + + with self.assertRaisesRegex(AssignmentHistoryIntegrityError, "runtime integrity"): + read_assignment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"allocation_ratio"}), + policy=self.policy, + read_port=FakeAssignmentHistoryPort((record,)), + ) + + def test_duplicate_visible_assignment_identity_fails_closed(self) -> None: + duplicate = assignment_record(effective_from=date(2026, 2, 1), effective_to=None) + with self.assertRaisesRegex(AssignmentHistoryIntegrityError, "duplicate visible assignment"): + read_assignment_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=FakeAssignmentHistoryPort((assignment_record(), duplicate)), + ) + + def test_policy_schema_drift_fails_closed_after_authorization(self) -> None: + drifted_policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="drifted-v1", + resource_kind="person_assignment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.assignment_history.read", + permitted_fields=frozenset({"future_sensitive_field"}), + ) + with self.assertRaisesRegex(AssignmentHistoryIntegrityError, "unsupported assignment-history field"): + read_assignment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({"future_sensitive_field"}), + policy=drifted_policy, + read_port=FakeAssignmentHistoryPort((assignment_record(),)), + ) + + def test_field_name_subclass_from_authorization_fails_closed(self) -> None: + forged_field = ForgedField("effective_from") + forged_policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="forged-field-v1", + resource_kind="person_assignment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.assignment_history.read", + permitted_fields=frozenset({forged_field}), + ) + + with self.assertRaisesRegex(AssignmentHistoryIntegrityError, "unsupported assignment-history field"): + read_assignment_history( + principal=self.principal, + tenant_record_id=TENANT, + person_record_id=PERSON, + known_at=KNOWN_AT, + purpose_code="employee_profile_review", + requested_fields=frozenset({forged_field}), + policy=forged_policy, + read_port=FakeAssignmentHistoryPort((assignment_record(),)), + ) + + def test_invalid_request_shape_fails_before_repository_access(self) -> None: + port = FakeAssignmentHistoryPort(()) + 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, 28, 3, 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, value_type=type(value).__name__), self.assertRaises(ValueError): + read_assignment_history(**kwargs) + self.assertEqual(port.calls, []) + + def test_record_rejects_noncanonical_business_or_time_values(self) -> None: + invalid_overrides = ( + {"assignment_record_id": "assignment"}, + {"assignment_record_id": ForgedUUID(str(ASSIGNMENT_A))}, + {"allocation_ratio": Decimal("NaN")}, + {"allocation_ratio": Decimal("1.00000")}, + {"allocation_ratio": Decimal("0.0000")}, + {"allocation_ratio": Decimal("1.0001")}, + {"effective_from": datetime(2026, 1, 1, tzinfo=timezone.utc)}, + {"effective_to": date(2026, 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): + assignment_record(**override) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_assignment_history_timezone_integrity.py b/services/people-api/tests/test_assignment_history_timezone_integrity.py new file mode 100644 index 000000000..b3cddb6a2 --- /dev/null +++ b/services/people-api/tests/test_assignment_history_timezone_integrity.py @@ -0,0 +1,119 @@ +"""Regression coverage for assignment-history timezone-provider integrity.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, tzinfo +from decimal import Decimal +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.assignment_history import AssignmentHistoryRecord, read_assignment_history + +TENANT = UUID("0198a412-7000-7000-8000-000000000001") +PERSON = UUID("0198a412-7000-7000-8000-000000000010") +EMPLOYMENT = UUID("0198a412-7000-7000-8000-000000000020") +POSITION = UUID("0198a412-7000-7000-8000-000000000030") +ASSIGNMENT = UUID("0198a412-7000-7000-8000-000000000040") + + +class CallerControlledUtc(tzinfo): + """Model a caller-defined timezone provider that currently reports UTC.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + """Return zero offset while retaining caller-controlled behavior.""" + del dt + return timedelta(0) + + def dst(self, dt: datetime | None) -> timedelta | None: + """Return no daylight-saving adjustment.""" + del dt + return None + + def tzname(self, dt: datetime | None) -> str: + """Present the forged provider as UTC-like text.""" + del dt + return "UTC" + + +class EmptyPort: + """Capture whether protected persistence was reached.""" + + def __init__(self) -> None: + """Start with no protected reads.""" + self.calls = 0 + + def read_assignment_history( + self, + *, + tenant_record_id: UUID, + person_record_id: UUID, + known_at: datetime, + ) -> tuple[AssignmentHistoryRecord, ...]: + """Record protected access and return no assignment rows.""" + del tenant_record_id, person_record_id, known_at + self.calls += 1 + return () + + +class AssignmentHistoryTimezoneIntegrityTests(unittest.TestCase): + """Prove trust-bearing instants cannot retain caller-controlled tzinfo behavior.""" + + def setUp(self) -> None: + """Build the minimum authorized read context.""" + self.principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse:hr-operator", + granted_scope_codes=frozenset({"orgmetra.people.assignment_history.read"}), + ) + self.policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="employee-profile-assignment-history-v1", + resource_kind="person_assignment_history", + purpose_code="employee_profile_review", + operation_code="read_record", + required_scope_code="orgmetra.people.assignment_history.read", + permitted_fields=frozenset({"effective_from"}), + ) + + def test_caller_controlled_known_at_fails_before_protected_retrieval(self) -> None: + """Reject a UTC-looking custom tzinfo before any repository call.""" + port = EmptyPort() + known_at = datetime(2026, 8, 28, 3, 0, tzinfo=CallerControlledUtc()) + + with self.assertRaisesRegex(ValueError, "known_at must be a timezone-aware UTC datetime"): + read_assignment_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=port, + ) + + self.assertEqual(port.calls, 0) + + def test_persisted_record_rejects_caller_controlled_recorded_timezone(self) -> None: + """Reject UTC-looking custom tzinfo before it becomes persisted read evidence.""" + recorded_from = datetime(2026, 8, 20, 0, 0, tzinfo=CallerControlledUtc()) + + with self.assertRaisesRegex(ValueError, "recorded_from must be a timezone-aware UTC datetime"): + AssignmentHistoryRecord( + tenant_record_id=TENANT, + assignment_record_id=ASSIGNMENT, + employment_record_id=EMPLOYMENT, + person_record_id=PERSON, + position_record_id=POSITION, + allocation_ratio=Decimal("1.0000"), + effective_from=datetime(2026, 1, 1).date(), + effective_to=None, + recorded_from=recorded_from, + recorded_to=None, + ) + + +if __name__ == "__main__": + unittest.main()