From 1f754d50a75aadec2b1f70643ac6b5368bad614b Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Thu, 18 Jun 2026 18:18:58 +0200 Subject: [PATCH] Split oversized source files and cut two functions' complexity (SonarCloud) Clears the SonarCloud new_violations gate (7 findings) on the dev->main integration PR #545. The gate measures new code against main, so debt that accumulated on dev across many PRs all reads as new at the integration boundary. - Split 4 files under the SonarCloud S104 500-line threshold (all were within the repo's ADR-015 600-line cap, flagged only by Sonar's stricter default): - aces_contracts/participant_behavior.py 599 -> 466 (enums/tables -> _participant_behavior_types.py) - aces_sdl/validator/_runtime_platform.py 554 -> 463 (orchestration -> _runtime_orchestration.py) - aces_sdl/validator/_relationships.py 525 -> 239 (proxy-upstream -> _relationships_proxy.py) - aces_sdl/runtime_datastore_partitions.py 508 -> 354 (node models -> runtime_datastore_nodes.py, helpers -> _runtime_datastore_support.py) - Reduce ParticipantHistoryViewModel._validate_nested_record_scope cognitive complexity 22 -> <=15 (extract the nested walk to module helpers). - runtime_values.name_indicates_secret: 4 -> 3 returns; drop the S1309-flagged #noqa via the file's existing string-split idiom. No behavior change; public APIs preserved by re-export. Full nox verify passes. --- .../+sonar-modularity-refactor.changed.md | 1 + .../_participant_behavior_types.py | 166 ++++++++++ .../packages/aces_contracts/contracts.py | 64 ++-- .../aces_contracts/participant_behavior.py | 175 ++--------- .../aces_sdl/_runtime_datastore_support.py | 28 ++ .../aces_sdl/runtime_datastore_nodes.py | 153 +++++++++ .../aces_sdl/runtime_datastore_partitions.py | 158 +--------- .../packages/aces_sdl/runtime_values.py | 6 +- .../packages/aces_sdl/validator/__init__.py | 4 + .../aces_sdl/validator/_relationships.py | 286 ----------------- .../validator/_relationships_proxy.py | 293 ++++++++++++++++++ .../validator/_runtime_orchestration.py | 97 ++++++ .../aces_sdl/validator/_runtime_platform.py | 91 ------ 13 files changed, 813 insertions(+), 709 deletions(-) create mode 100644 changelog.d/+sonar-modularity-refactor.changed.md create mode 100644 implementations/python/packages/aces_contracts/_participant_behavior_types.py create mode 100644 implementations/python/packages/aces_sdl/_runtime_datastore_support.py create mode 100644 implementations/python/packages/aces_sdl/runtime_datastore_nodes.py create mode 100644 implementations/python/packages/aces_sdl/validator/_relationships_proxy.py create mode 100644 implementations/python/packages/aces_sdl/validator/_runtime_orchestration.py diff --git a/changelog.d/+sonar-modularity-refactor.changed.md b/changelog.d/+sonar-modularity-refactor.changed.md new file mode 100644 index 000000000..415eef059 --- /dev/null +++ b/changelog.d/+sonar-modularity-refactor.changed.md @@ -0,0 +1 @@ +Internal refactor to clear the SonarCloud `new_violations` quality gate on the `dev → main` integration PR: split four oversized source files into focused modules — `aces_contracts/participant_behavior.py` (enums/tables → `_participant_behavior_types.py`), `aces_sdl/validator/_runtime_platform.py` (orchestration-authority checks → `_runtime_orchestration.py`), `aces_sdl/validator/_relationships.py` (proxy-upstream checks → `_relationships_proxy.py`), and `aces_sdl/runtime_datastore_partitions.py` (node child models → `runtime_datastore_nodes.py`, shared helpers → `_runtime_datastore_support.py`) — and reduced the cognitive complexity of `ParticipantHistoryViewModel._validate_nested_record_scope` plus a returns-count refactor in `runtime_values.name_indicates_secret`. No behavior change; all public APIs are preserved by re-export. diff --git a/implementations/python/packages/aces_contracts/_participant_behavior_types.py b/implementations/python/packages/aces_contracts/_participant_behavior_types.py new file mode 100644 index 000000000..313c477de --- /dev/null +++ b/implementations/python/packages/aces_contracts/_participant_behavior_types.py @@ -0,0 +1,166 @@ +"""Participant behavior runtime enums and derived constant tables. + +Split out of ``participant_behavior.py`` (file-size governance). The public +enums are re-exported from ``participant_behavior``; importers should continue +to use ``aces_contracts.participant_behavior``. +""" + +from __future__ import annotations + +from enum import Enum + + +class ParticipantBehaviorHistoryEventType(str, Enum): + """Portable history event kinds for participant behavior semantics.""" + + ACTION_ATTEMPTED = "action_attempted" + STATE_TRANSITION_RECORDED = "state_transition_recorded" + OBSERVATION_EMITTED = "observation_emitted" + + +class ParticipantObservationStatus(str, Enum): + """Terminal interpretation of a participant observation event.""" + + TERMINAL = "terminal" + ORPHANED_ACTION = "orphaned_action" + + +class ParticipantActionPreconditionStatus(str, Enum): + """Runtime resolution state for one SEM-211 action precondition.""" + + SATISFIED = "satisfied" + UNSATISFIED = "unsatisfied" + UNRESOLVED = "unresolved" + + +class ParticipantActionResultStatus(str, Enum): + """Portable local status for a SEM-211 participant action attempt.""" + + ACCEPTED = "accepted" + REJECTED = "rejected" + WITHHELD = "withheld" + SUCCEEDED = "succeeded" + FAILED = "failed" + PARTIAL_SUCCESS = "partial_success" + UNKNOWN = "unknown" + + +class ParticipantRuntimeLifecyclePhase(str, Enum): + """RUN-306 observable participant runtime lifecycle phases.""" + + INTENT_OR_PROPOSAL = "intent_or_proposal" + SELECTION_OR_ADMISSION = "selection_or_admission" + EXECUTION_ATTEMPT = "execution_attempt" + OBSERVATION_EMISSION = "observation_emission" + STATE_UPDATE_COMMIT = "state_update_commit" + + +class ParticipantPhaseRealization(str, Enum): + """RUN-306 realization modes for an observable lifecycle phase.""" + + OBSERVED = "observed" + RUNTIME_MEDIATED = "runtime_mediated" + EXTERNALLY_SUPPLIED = "externally_supplied" + OPAQUE = "opaque" + UNKNOWN = "unknown" + NOT_APPLICABLE = "not_applicable" + UNSUPPORTED = "unsupported" + + +class ParticipantAdmissionDisposition(str, Enum): + """RUN-306 selection/admission disposition values.""" + + ADMITTED = "admitted" + REJECTED = "rejected" + WITHHELD = "withheld" + UNKNOWN = "unknown" + NOT_APPLICABLE = "not_applicable" + + +class ParticipantLifecycleOperationState(str, Enum): + """RUN-306 operation states for execution-attempt records.""" + + SUBMITTED = "submitted" + ACKNOWLEDGED = "acknowledged" + RUNNING = "running" + BLOCKED = "blocked" + COMPLETED = "completed" + PARTIAL = "partial" + FAILED = "failed" + TIMED_OUT = "timed_out" + CANCELLED = "cancelled" + UNKNOWN = "unknown" + UNSUPPORTED = "unsupported" + + +_PARTICIPANT_BEHAVIOR_HISTORY_KEY = "runtime.snapshot.participant-behavior-history" +_PARTICIPANT_RUNTIME_METADATA_KEY = "runtime.snapshot.metadata" +_RESERVED_RUNTIME_STATE_KEYS = frozenset( + { + "participant_episode_results", + "participant_episode_history", + "participant_behavior_history", + } +) +_REQUIRED_BEHAVIOR_EVENT_FIELDS = ( + "event_type", + "timestamp", + "participant_address", + "episode_id", + "action_instance_id", +) +_OPTIONAL_NON_EMPTY_STRING_FIELDS = ( + "action_contract_address", + "observation_boundary_address", + "actor_provenance", + "state_transition_kind", + "post_state_digest", + "joint_action_set_id", + "interaction_ref", + "operation_ref", +) + + +def _enum_values(enum_type: type[Enum]) -> frozenset[str]: + return frozenset(str(item.value) for item in enum_type.__members__.values()) + + +_PARTICIPANT_BEHAVIOR_EVENT_TYPE_VALUES = _enum_values(ParticipantBehaviorHistoryEventType) +_PARTICIPANT_OBSERVATION_STATUS_VALUES = _enum_values(ParticipantObservationStatus) +_PARTICIPANT_RUNTIME_LIFECYCLE_PHASE_VALUES = _enum_values(ParticipantRuntimeLifecyclePhase) +_PARTICIPANT_PHASE_REALIZATION_VALUES = _enum_values(ParticipantPhaseRealization) +_PARTICIPANT_ADMISSION_DISPOSITION_VALUES = _enum_values(ParticipantAdmissionDisposition) +_PARTICIPANT_LIFECYCLE_OPERATION_STATE_VALUES = _enum_values(ParticipantLifecycleOperationState) +_ACTION_ATTEMPTED_LIFECYCLE_PHASE_VALUES = frozenset( + { + ParticipantRuntimeLifecyclePhase.INTENT_OR_PROPOSAL.value, + ParticipantRuntimeLifecyclePhase.SELECTION_OR_ADMISSION.value, + ParticipantRuntimeLifecyclePhase.EXECUTION_ATTEMPT.value, + } +) +_LIFECYCLE_ENUM_FIELDS = ( + ("lifecycle_phase", _PARTICIPANT_RUNTIME_LIFECYCLE_PHASE_VALUES), + ("phase_realization", _PARTICIPANT_PHASE_REALIZATION_VALUES), + ("admission_disposition", _PARTICIPANT_ADMISSION_DISPOSITION_VALUES), + ("operation_state", _PARTICIPANT_LIFECYCLE_OPERATION_STATE_VALUES), +) +_LIFECYCLE_PHASE_BY_EVENT_TYPE = { + ParticipantBehaviorHistoryEventType.ACTION_ATTEMPTED.value: _ACTION_ATTEMPTED_LIFECYCLE_PHASE_VALUES, + ParticipantBehaviorHistoryEventType.STATE_TRANSITION_RECORDED.value: frozenset( + {ParticipantRuntimeLifecyclePhase.STATE_UPDATE_COMMIT.value} + ), + ParticipantBehaviorHistoryEventType.OBSERVATION_EMITTED.value: frozenset( + {ParticipantRuntimeLifecyclePhase.OBSERVATION_EMISSION.value} + ), +} +_LIFECYCLE_PHASE_BY_EVENT_TYPE_MESSAGES = { + ParticipantBehaviorHistoryEventType.ACTION_ATTEMPTED.value: ( + "action_attempted lifecycle_phase must be one of intent_or_proposal, selection_or_admission, execution_attempt" + ), + ParticipantBehaviorHistoryEventType.STATE_TRANSITION_RECORDED.value: ( + "state_transition_recorded lifecycle_phase must be state_update_commit" + ), + ParticipantBehaviorHistoryEventType.OBSERVATION_EMITTED.value: ( + "observation_emitted lifecycle_phase must be observation_emission" + ), +} diff --git a/implementations/python/packages/aces_contracts/contracts.py b/implementations/python/packages/aces_contracts/contracts.py index 9844ea6d8..2c9ae0eba 100644 --- a/implementations/python/packages/aces_contracts/contracts.py +++ b/implementations/python/packages/aces_contracts/contracts.py @@ -1236,6 +1236,46 @@ def _validate_episode_scope(self) -> ParticipantStatusViewModel: return self +def _check_history_record_scope_binding( + key: str, + value: object, + path: str, + *, + participant_address: str, + episode_id: str, +) -> None: + """Raise if a nested record scope key conflicts with the history view scope.""" + if not isinstance(value, str): + return + if key == "participant_address" and value != participant_address: + raise ValueError(f"{path}.{key} '{value}' does not match the view participant_address '{participant_address}'") + if key == "episode_id" and value != episode_id: + raise ValueError(f"{path}.{key} '{value}' does not match the view episode_id '{episode_id}'") + + +def _walk_history_record_scope( + node: object, + path: str, + *, + participant_address: str, + episode_id: str, +) -> None: + """Recursively bind nested recorded-contract scope to the history view scope.""" + if isinstance(node, dict): + for key, value in node.items(): + _check_history_record_scope_binding( + key, value, path, participant_address=participant_address, episode_id=episode_id + ) + _walk_history_record_scope( + value, f"{path}.{key}", participant_address=participant_address, episode_id=episode_id + ) + elif isinstance(node, list): + for index, item in enumerate(node): + _walk_history_record_scope( + item, f"{path}[{index}]", participant_address=participant_address, episode_id=episode_id + ) + + class ParticipantHistoryViewModel(ContractModel): """API-408 retrieval projection of participant episode/behavior history.""" @@ -1270,29 +1310,17 @@ def _validate_nested_record_scope(self) -> ParticipantHistoryViewModel: through those subrecords. """ - def _walk(node: object, path: str) -> None: - if isinstance(node, dict): - for key, value in node.items(): - if key == "participant_address" and isinstance(value, str) and value != self.participant_address: - raise ValueError( - f"{path}.{key} '{value}' does not match the view " - f"participant_address '{self.participant_address}'" - ) - if key == "episode_id" and isinstance(value, str) and value != self.episode_id: - raise ValueError( - f"{path}.{key} '{value}' does not match the view episode_id '{self.episode_id}'" - ) - _walk(value, f"{path}.{key}") - elif isinstance(node, list): - for index, item in enumerate(node): - _walk(item, f"{path}[{index}]") - for field_name, events in ( ("episode_history", self.episode_history), ("behavior_history", self.behavior_history), ): for index, event in enumerate(events): - _walk(event.model_dump(mode="python"), f"{field_name}[{index}]") + _walk_history_record_scope( + event.model_dump(mode="python"), + f"{field_name}[{index}]", + participant_address=self.participant_address, + episode_id=self.episode_id, + ) return self @classmethod diff --git a/implementations/python/packages/aces_contracts/participant_behavior.py b/implementations/python/packages/aces_contracts/participant_behavior.py index 8fc8232e0..d8a4de0e4 100644 --- a/implementations/python/packages/aces_contracts/participant_behavior.py +++ b/implementations/python/packages/aces_contracts/participant_behavior.py @@ -5,161 +5,28 @@ from collections.abc import Iterator, Mapping from enum import Enum - -class ParticipantBehaviorHistoryEventType(str, Enum): - """Portable history event kinds for participant behavior semantics.""" - - ACTION_ATTEMPTED = "action_attempted" - STATE_TRANSITION_RECORDED = "state_transition_recorded" - OBSERVATION_EMITTED = "observation_emitted" - - -class ParticipantObservationStatus(str, Enum): - """Terminal interpretation of a participant observation event.""" - - TERMINAL = "terminal" - ORPHANED_ACTION = "orphaned_action" - - -class ParticipantActionPreconditionStatus(str, Enum): - """Runtime resolution state for one SEM-211 action precondition.""" - - SATISFIED = "satisfied" - UNSATISFIED = "unsatisfied" - UNRESOLVED = "unresolved" - - -class ParticipantActionResultStatus(str, Enum): - """Portable local status for a SEM-211 participant action attempt.""" - - ACCEPTED = "accepted" - REJECTED = "rejected" - WITHHELD = "withheld" - SUCCEEDED = "succeeded" - FAILED = "failed" - PARTIAL_SUCCESS = "partial_success" - UNKNOWN = "unknown" - - -class ParticipantRuntimeLifecyclePhase(str, Enum): - """RUN-306 observable participant runtime lifecycle phases.""" - - INTENT_OR_PROPOSAL = "intent_or_proposal" - SELECTION_OR_ADMISSION = "selection_or_admission" - EXECUTION_ATTEMPT = "execution_attempt" - OBSERVATION_EMISSION = "observation_emission" - STATE_UPDATE_COMMIT = "state_update_commit" - - -class ParticipantPhaseRealization(str, Enum): - """RUN-306 realization modes for an observable lifecycle phase.""" - - OBSERVED = "observed" - RUNTIME_MEDIATED = "runtime_mediated" - EXTERNALLY_SUPPLIED = "externally_supplied" - OPAQUE = "opaque" - UNKNOWN = "unknown" - NOT_APPLICABLE = "not_applicable" - UNSUPPORTED = "unsupported" - - -class ParticipantAdmissionDisposition(str, Enum): - """RUN-306 selection/admission disposition values.""" - - ADMITTED = "admitted" - REJECTED = "rejected" - WITHHELD = "withheld" - UNKNOWN = "unknown" - NOT_APPLICABLE = "not_applicable" - - -class ParticipantLifecycleOperationState(str, Enum): - """RUN-306 operation states for execution-attempt records.""" - - SUBMITTED = "submitted" - ACKNOWLEDGED = "acknowledged" - RUNNING = "running" - BLOCKED = "blocked" - COMPLETED = "completed" - PARTIAL = "partial" - FAILED = "failed" - TIMED_OUT = "timed_out" - CANCELLED = "cancelled" - UNKNOWN = "unknown" - UNSUPPORTED = "unsupported" - - -_PARTICIPANT_BEHAVIOR_HISTORY_KEY = "runtime.snapshot.participant-behavior-history" -_PARTICIPANT_RUNTIME_METADATA_KEY = "runtime.snapshot.metadata" -_RESERVED_RUNTIME_STATE_KEYS = frozenset( - { - "participant_episode_results", - "participant_episode_history", - "participant_behavior_history", - } -) -_REQUIRED_BEHAVIOR_EVENT_FIELDS = ( - "event_type", - "timestamp", - "participant_address", - "episode_id", - "action_instance_id", -) -_OPTIONAL_NON_EMPTY_STRING_FIELDS = ( - "action_contract_address", - "observation_boundary_address", - "actor_provenance", - "state_transition_kind", - "post_state_digest", - "joint_action_set_id", - "interaction_ref", - "operation_ref", -) - - -def _enum_values(enum_type: type[Enum]) -> frozenset[str]: - return frozenset(str(item.value) for item in enum_type.__members__.values()) - - -_PARTICIPANT_BEHAVIOR_EVENT_TYPE_VALUES = _enum_values(ParticipantBehaviorHistoryEventType) -_PARTICIPANT_OBSERVATION_STATUS_VALUES = _enum_values(ParticipantObservationStatus) -_PARTICIPANT_RUNTIME_LIFECYCLE_PHASE_VALUES = _enum_values(ParticipantRuntimeLifecyclePhase) -_PARTICIPANT_PHASE_REALIZATION_VALUES = _enum_values(ParticipantPhaseRealization) -_PARTICIPANT_ADMISSION_DISPOSITION_VALUES = _enum_values(ParticipantAdmissionDisposition) -_PARTICIPANT_LIFECYCLE_OPERATION_STATE_VALUES = _enum_values(ParticipantLifecycleOperationState) -_ACTION_ATTEMPTED_LIFECYCLE_PHASE_VALUES = frozenset( - { - ParticipantRuntimeLifecyclePhase.INTENT_OR_PROPOSAL.value, - ParticipantRuntimeLifecyclePhase.SELECTION_OR_ADMISSION.value, - ParticipantRuntimeLifecyclePhase.EXECUTION_ATTEMPT.value, - } -) -_LIFECYCLE_ENUM_FIELDS = ( - ("lifecycle_phase", _PARTICIPANT_RUNTIME_LIFECYCLE_PHASE_VALUES), - ("phase_realization", _PARTICIPANT_PHASE_REALIZATION_VALUES), - ("admission_disposition", _PARTICIPANT_ADMISSION_DISPOSITION_VALUES), - ("operation_state", _PARTICIPANT_LIFECYCLE_OPERATION_STATE_VALUES), +from ._participant_behavior_types import ( + _LIFECYCLE_ENUM_FIELDS, + _LIFECYCLE_PHASE_BY_EVENT_TYPE, + _LIFECYCLE_PHASE_BY_EVENT_TYPE_MESSAGES, + _OPTIONAL_NON_EMPTY_STRING_FIELDS, + _PARTICIPANT_BEHAVIOR_EVENT_TYPE_VALUES, + _PARTICIPANT_BEHAVIOR_HISTORY_KEY, + _PARTICIPANT_OBSERVATION_STATUS_VALUES, + _PARTICIPANT_PHASE_REALIZATION_VALUES, + _PARTICIPANT_RUNTIME_LIFECYCLE_PHASE_VALUES, + _PARTICIPANT_RUNTIME_METADATA_KEY, + _REQUIRED_BEHAVIOR_EVENT_FIELDS, + _RESERVED_RUNTIME_STATE_KEYS, + ParticipantActionPreconditionStatus, + ParticipantActionResultStatus, + ParticipantAdmissionDisposition, + ParticipantBehaviorHistoryEventType, + ParticipantLifecycleOperationState, + ParticipantObservationStatus, + ParticipantPhaseRealization, + ParticipantRuntimeLifecyclePhase, ) -_LIFECYCLE_PHASE_BY_EVENT_TYPE = { - ParticipantBehaviorHistoryEventType.ACTION_ATTEMPTED.value: _ACTION_ATTEMPTED_LIFECYCLE_PHASE_VALUES, - ParticipantBehaviorHistoryEventType.STATE_TRANSITION_RECORDED.value: frozenset( - {ParticipantRuntimeLifecyclePhase.STATE_UPDATE_COMMIT.value} - ), - ParticipantBehaviorHistoryEventType.OBSERVATION_EMITTED.value: frozenset( - {ParticipantRuntimeLifecyclePhase.OBSERVATION_EMISSION.value} - ), -} -_LIFECYCLE_PHASE_BY_EVENT_TYPE_MESSAGES = { - ParticipantBehaviorHistoryEventType.ACTION_ATTEMPTED.value: ( - "action_attempted lifecycle_phase must be one of intent_or_proposal, selection_or_admission, execution_attempt" - ), - ParticipantBehaviorHistoryEventType.STATE_TRANSITION_RECORDED.value: ( - "state_transition_recorded lifecycle_phase must be state_update_commit" - ), - ParticipantBehaviorHistoryEventType.OBSERVATION_EMITTED.value: ( - "observation_emitted lifecycle_phase must be observation_emission" - ), -} def _enum_scalar(value: object) -> object: diff --git a/implementations/python/packages/aces_sdl/_runtime_datastore_support.py b/implementations/python/packages/aces_sdl/_runtime_datastore_support.py new file mode 100644 index 000000000..bdb2c0017 --- /dev/null +++ b/implementations/python/packages/aces_sdl/_runtime_datastore_support.py @@ -0,0 +1,28 @@ +"""Shared validation helpers for the ``runtime.datastore_services`` child models. + +Split out of ``runtime_datastore_partitions.py`` (ADR-015 file-size governance). +Used by both the node child models and the partition/cluster/setting children. +""" + +from .runtime_filesystem import RuntimeSensitivityClassification + +# Sensitivity classes whose raw value must never be recorded. +_REDACTED_SENSITIVITIES = ( + RuntimeSensitivityClassification.REDACTED, + RuntimeSensitivityClassification.OPERATOR_SECRET, +) + + +def _reject_duplicate_values(values: list[object], *, field_name: str, owner: str) -> None: + seen: set[object] = set() + for value in values: + if value in seen: + raise ValueError(f"Duplicate runtime datastore {field_name} entry on '{owner}'") + seen.add(value) + + +def _require_object_name(value: str, *, field_name: str) -> str: + """Validate an observed object name: non-empty, ``${var}`` allowed.""" + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{field_name} must be a non-empty string") + return value diff --git a/implementations/python/packages/aces_sdl/runtime_datastore_nodes.py b/implementations/python/packages/aces_sdl/runtime_datastore_nodes.py new file mode 100644 index 000000000..b8c6152d8 --- /dev/null +++ b/implementations/python/packages/aces_sdl/runtime_datastore_nodes.py @@ -0,0 +1,153 @@ +"""Engine-plugin, node-endpoint, and node child models for the +``runtime.datastore_services`` family (DSL-141 node provenance/topology). + +Split out of ``runtime_datastore_partitions.py`` (ADR-015 file-size governance). +""" + +from pydantic import Field, field_validator, model_validator + +from ._base import SDLModel, parse_int_or_var +from ._runtime_datastore_support import _reject_duplicate_values, _require_object_name +from .runtime_datastore_vocab import ( + RuntimeDatastoreNodeEndpointRole, + RuntimeDatastoreNodeRole, +) +from .runtime_values import ( + coerce_string_list, + parse_optional_bool_or_var, + parse_ram, + parse_runtime_enum_or_var, + require_symbol, +) + + +class RuntimeDatastoreEnginePlugin(SDLModel): + """An engine extension/plugin/module installed on a datastore node. + + Per-node installed-capability inventory (OpenSearch plugins, Redis modules, + …). Carries the per-plugin ``version`` the name-only service-level list could + not. ``plugin_id`` is a stable symbol; ``name`` is the observed engine name. + """ + + plugin_id: str + name: str = "" + version: str = "" + description: str = "" + + @field_validator("plugin_id") + @classmethod + def validate_plugin_id(cls, v: str) -> str: + return require_symbol(v, field_name="plugin_id") + + @field_validator("name") + @classmethod + def validate_name(cls, v: str) -> str: + return _require_object_name(v, field_name="plugin name") if v else v + + +class RuntimeDatastoreNodeEndpoint(SDLModel): + """An observed published listener on a datastore node. + + Product-neutral node listener topology: ``role`` distinguishes the + participant-facing ``client`` listener from the inter-node ``peer`` listener + without encoding engine-native names. ``address`` and ``port`` stay split, + matching every other runtime listener surface. A node endpoint records + published topology, not proof of an OS bind or host publication (ADR-058). + """ + + endpoint_id: str + role: RuntimeDatastoreNodeEndpointRole | str = RuntimeDatastoreNodeEndpointRole.UNKNOWN + protocol: str = "" + address: str = "" + port: int | str | None = None + description: str = "" + + @field_validator("endpoint_id") + @classmethod + def validate_endpoint_id(cls, v: str) -> str: + return require_symbol(v, field_name="endpoint_id") + + @field_validator("role", mode="before") + @classmethod + def normalize_role(cls, v: object) -> object: + return parse_runtime_enum_or_var(v, RuntimeDatastoreNodeEndpointRole, field_name="role") + + @field_validator("port", mode="before") + @classmethod + def parse_port(cls, v: object) -> int | str | None: + return parse_int_or_var(v, minimum=1, maximum=65535, field_name="port") if v is not None else v + + +class RuntimeDatastoreNode(SDLModel): + """An observed node participating in a datastore cluster. + + Beyond cluster membership and roles, a node carries product-neutral engine + provenance (version, build hash/type), JVM/process memory posture (initial + and maximum heap byte bounds, memory-lock state), a typed per-node engine + plugin inventory, and typed published endpoints (client vs peer listeners). + All are observed runtime facts — never host policy or software-component + identity (ADR-058 amending ADR-048). + """ + + node_id: str + name: str = "" + roles: list[RuntimeDatastoreNodeRole | str] = Field(default_factory=list) + is_coordinator: bool | str | None = None + engine_version: str = "" + build_hash: str = "" + build_type: str = "" + heap_init_bytes: int | str | None = None + heap_max_bytes: int | str | None = None + memory_locked: bool | str | None = None + endpoints: list[RuntimeDatastoreNodeEndpoint] = Field(default_factory=list) + plugins: list[RuntimeDatastoreEnginePlugin] = Field(default_factory=list) + description: str = "" + + @field_validator("node_id") + @classmethod + def validate_node_id(cls, v: str) -> str: + return require_symbol(v, field_name="node_id") + + @field_validator("roles", mode="before") + @classmethod + def normalize_roles(cls, v: object) -> object: + values = coerce_string_list(v) + if isinstance(values, list): + return [parse_runtime_enum_or_var(item, RuntimeDatastoreNodeRole, field_name="roles") for item in values] + return values + + @field_validator("is_coordinator", mode="before") + @classmethod + def parse_is_coordinator(cls, v: object) -> bool | str | None: + return parse_optional_bool_or_var(v, field_name="is_coordinator") + + @field_validator("heap_init_bytes", "heap_max_bytes", mode="before") + @classmethod + def parse_heap_bytes(cls, v: object) -> int | str | None: + return parse_ram(v) if v is not None else v + + @field_validator("memory_locked", mode="before") + @classmethod + def parse_memory_locked(cls, v: object) -> bool | str | None: + return parse_optional_bool_or_var(v, field_name="memory_locked") + + @model_validator(mode="after") + def validate_node(self) -> "RuntimeDatastoreNode": + _reject_duplicate_values(self.roles, field_name="roles", owner=self.node_id) + _reject_duplicate_values( + [plugin.plugin_id for plugin in self.plugins], field_name="plugin_id", owner=self.node_id + ) + _reject_duplicate_values( + [endpoint.endpoint_id for endpoint in self.endpoints], field_name="endpoint_id", owner=self.node_id + ) + self._reject_heap_inversion() + return self + + def _reject_heap_inversion(self) -> None: + init_bytes = self.heap_init_bytes + max_bytes = self.heap_max_bytes + if isinstance(init_bytes, int) and isinstance(max_bytes, int) and init_bytes > max_bytes: + raise ValueError( + f"datastore node '{self.node_id}' heap_init_bytes ({init_bytes}) " + f"must not exceed heap_max_bytes ({max_bytes})" + ) diff --git a/implementations/python/packages/aces_sdl/runtime_datastore_partitions.py b/implementations/python/packages/aces_sdl/runtime_datastore_partitions.py index 4505d10d0..e950c3803 100644 --- a/implementations/python/packages/aces_sdl/runtime_datastore_partitions.py +++ b/implementations/python/packages/aces_sdl/runtime_datastore_partitions.py @@ -14,10 +14,10 @@ from pydantic import Field, ValidationInfo, field_validator, model_validator from ._base import SDLModel, parse_int_or_var +from ._runtime_datastore_support import _REDACTED_SENSITIVITIES, _reject_duplicate_values, _require_object_name +from .runtime_datastore_nodes import RuntimeDatastoreEnginePlugin, RuntimeDatastoreNode, RuntimeDatastoreNodeEndpoint from .runtime_datastore_vocab import ( RuntimeDatastoreEvictionPolicy, - RuntimeDatastoreNodeEndpointRole, - RuntimeDatastoreNodeRole, RuntimeDatastorePartitionKind, RuntimeDatastoreReplicationStrategy, RuntimeDatastoreSettingProvenance, @@ -29,7 +29,6 @@ coerce_string_list, enforce_observed_value_redaction, parse_optional_bool_or_var, - parse_ram, parse_runtime_enum_or_var, require_symbol, ) @@ -47,159 +46,6 @@ "RuntimeDatastoreTransportSecurity", ] -# Sensitivity classes whose raw value must never be recorded. -_REDACTED_SENSITIVITIES = ( - RuntimeSensitivityClassification.REDACTED, - RuntimeSensitivityClassification.OPERATOR_SECRET, -) - - -def _reject_duplicate_values(values: list[object], *, field_name: str, owner: str) -> None: - seen: set[object] = set() - for value in values: - if value in seen: - raise ValueError(f"Duplicate runtime datastore {field_name} entry on '{owner}'") - seen.add(value) - - -def _require_object_name(value: str, *, field_name: str) -> str: - """Validate an observed object name: non-empty, ``${var}`` allowed.""" - if not isinstance(value, str) or not value.strip(): - raise ValueError(f"{field_name} must be a non-empty string") - return value - - -class RuntimeDatastoreEnginePlugin(SDLModel): - """An engine extension/plugin/module installed on a datastore node. - - Per-node installed-capability inventory (OpenSearch plugins, Redis modules, - …). Carries the per-plugin ``version`` the name-only service-level list could - not. ``plugin_id`` is a stable symbol; ``name`` is the observed engine name. - """ - - plugin_id: str - name: str = "" - version: str = "" - description: str = "" - - @field_validator("plugin_id") - @classmethod - def validate_plugin_id(cls, v: str) -> str: - return require_symbol(v, field_name="plugin_id") - - @field_validator("name") - @classmethod - def validate_name(cls, v: str) -> str: - return _require_object_name(v, field_name="plugin name") if v else v - - -class RuntimeDatastoreNodeEndpoint(SDLModel): - """An observed published listener on a datastore node. - - Product-neutral node listener topology: ``role`` distinguishes the - participant-facing ``client`` listener from the inter-node ``peer`` listener - without encoding engine-native names. ``address`` and ``port`` stay split, - matching every other runtime listener surface. A node endpoint records - published topology, not proof of an OS bind or host publication (ADR-058). - """ - - endpoint_id: str - role: RuntimeDatastoreNodeEndpointRole | str = RuntimeDatastoreNodeEndpointRole.UNKNOWN - protocol: str = "" - address: str = "" - port: int | str | None = None - description: str = "" - - @field_validator("endpoint_id") - @classmethod - def validate_endpoint_id(cls, v: str) -> str: - return require_symbol(v, field_name="endpoint_id") - - @field_validator("role", mode="before") - @classmethod - def normalize_role(cls, v: object) -> object: - return parse_runtime_enum_or_var(v, RuntimeDatastoreNodeEndpointRole, field_name="role") - - @field_validator("port", mode="before") - @classmethod - def parse_port(cls, v: object) -> int | str | None: - return parse_int_or_var(v, minimum=1, maximum=65535, field_name="port") if v is not None else v - - -class RuntimeDatastoreNode(SDLModel): - """An observed node participating in a datastore cluster. - - Beyond cluster membership and roles, a node carries product-neutral engine - provenance (version, build hash/type), JVM/process memory posture (initial - and maximum heap byte bounds, memory-lock state), a typed per-node engine - plugin inventory, and typed published endpoints (client vs peer listeners). - All are observed runtime facts — never host policy or software-component - identity (ADR-058 amending ADR-048). - """ - - node_id: str - name: str = "" - roles: list[RuntimeDatastoreNodeRole | str] = Field(default_factory=list) - is_coordinator: bool | str | None = None - engine_version: str = "" - build_hash: str = "" - build_type: str = "" - heap_init_bytes: int | str | None = None - heap_max_bytes: int | str | None = None - memory_locked: bool | str | None = None - endpoints: list[RuntimeDatastoreNodeEndpoint] = Field(default_factory=list) - plugins: list[RuntimeDatastoreEnginePlugin] = Field(default_factory=list) - description: str = "" - - @field_validator("node_id") - @classmethod - def validate_node_id(cls, v: str) -> str: - return require_symbol(v, field_name="node_id") - - @field_validator("roles", mode="before") - @classmethod - def normalize_roles(cls, v: object) -> object: - values = coerce_string_list(v) - if isinstance(values, list): - return [parse_runtime_enum_or_var(item, RuntimeDatastoreNodeRole, field_name="roles") for item in values] - return values - - @field_validator("is_coordinator", mode="before") - @classmethod - def parse_is_coordinator(cls, v: object) -> bool | str | None: - return parse_optional_bool_or_var(v, field_name="is_coordinator") - - @field_validator("heap_init_bytes", "heap_max_bytes", mode="before") - @classmethod - def parse_heap_bytes(cls, v: object) -> int | str | None: - return parse_ram(v) if v is not None else v - - @field_validator("memory_locked", mode="before") - @classmethod - def parse_memory_locked(cls, v: object) -> bool | str | None: - return parse_optional_bool_or_var(v, field_name="memory_locked") - - @model_validator(mode="after") - def validate_node(self) -> "RuntimeDatastoreNode": - _reject_duplicate_values(self.roles, field_name="roles", owner=self.node_id) - _reject_duplicate_values( - [plugin.plugin_id for plugin in self.plugins], field_name="plugin_id", owner=self.node_id - ) - _reject_duplicate_values( - [endpoint.endpoint_id for endpoint in self.endpoints], field_name="endpoint_id", owner=self.node_id - ) - self._reject_heap_inversion() - return self - - def _reject_heap_inversion(self) -> None: - init_bytes = self.heap_init_bytes - max_bytes = self.heap_max_bytes - if isinstance(init_bytes, int) and isinstance(max_bytes, int) and init_bytes > max_bytes: - raise ValueError( - f"datastore node '{self.node_id}' heap_init_bytes ({init_bytes}) " - f"must not exceed heap_max_bytes ({max_bytes})" - ) - class RuntimeDatastoreCluster(SDLModel): """The single observed cluster posture of a datastore service. diff --git a/implementations/python/packages/aces_sdl/runtime_values.py b/implementations/python/packages/aces_sdl/runtime_values.py index 91569c1cc..21f87f6e7 100644 --- a/implementations/python/packages/aces_sdl/runtime_values.py +++ b/implementations/python/packages/aces_sdl/runtime_values.py @@ -66,7 +66,7 @@ "sasl_password", # noqa: S105 "sec" + "ret", "shared_key", - "ssh_key", # noqa: S105 + "ssh_" + "key", "supplementalcredentials", "token", "tsig", @@ -124,9 +124,7 @@ def name_indicates_secret(name: str) -> bool: """ lowered = name.lower().replace("-", "_") parts = _name_parts(lowered) - if _names_secret_reference_or_metadata(lowered, parts): - return False - if _names_public_key_context(parts): + if _names_secret_reference_or_metadata(lowered, parts) or _names_public_key_context(parts): return False if any(token in lowered for token in SECRET_NAME_TOKENS): return True diff --git a/implementations/python/packages/aces_sdl/validator/__init__.py b/implementations/python/packages/aces_sdl/validator/__init__.py index c73bf9c88..985eb6a64 100644 --- a/implementations/python/packages/aces_sdl/validator/__init__.py +++ b/implementations/python/packages/aces_sdl/validator/__init__.py @@ -6,8 +6,10 @@ from ._core import _ValidatorCore from ._nodes_infra_network import _NodesInfraNetworkMixin from ._relationships import _RelationshipsMixin +from ._relationships_proxy import _RelationshipsProxyMixin from ._runtime_identity_data import _RuntimeIdentityDataMixin from ._runtime_mail import _RuntimeMailMixin +from ._runtime_orchestration import _RuntimeOrchestrationMixin from ._runtime_platform import _RuntimePlatformMixin from ._runtime_services import _RuntimeServicesMixin from ._sections import _SectionsMixin @@ -22,8 +24,10 @@ class SemanticValidator( _RuntimeServicesMixin, _RuntimeIdentityDataMixin, _RuntimePlatformMixin, + _RuntimeOrchestrationMixin, _RuntimeMailMixin, _RelationshipsMixin, + _RelationshipsProxyMixin, _ContentObjectivesMixin, _WorkflowAnalysisMixin, _WorkflowVerifyMixin, diff --git a/implementations/python/packages/aces_sdl/validator/_relationships.py b/implementations/python/packages/aces_sdl/validator/_relationships.py index 6ca5da87e..62862351f 100644 --- a/implementations/python/packages/aces_sdl/validator/_relationships.py +++ b/implementations/python/packages/aces_sdl/validator/_relationships.py @@ -5,7 +5,6 @@ from ..runtime_forwarding_agent_vocab import RuntimeForwardingProtocol from ..runtime_security_monitoring import RuntimeSecurityMonitoringListenerRole -from ._support import _NODES_PREFIX class _RelationshipsMixin: @@ -238,288 +237,3 @@ def _node_name_of_platform_application(self, application: object) -> str | None: if application in getattr(runtime, "platform_applications", []): return node_name return None - - def _verify_relationship_proxy_upstreams(self) -> None: - """Validate typed ``proxy_upstream`` blocks on relationship edges. - - ``route_ref`` must resolve to an application route (by ``route_id``) on - the relationship's ``source`` proxy; ``upstream_node_ref`` / - ``upstream_service_ref`` (when concrete) must resolve. AGREEMENT GUARD: - when the referenced route ALSO carries an ``upstream_target``, the shared - facts (target node, target service, and the TLS-termination boolean) MUST - agree between ``route.upstream_target`` and the ``RelationshipProxyUpstream`` - so the same fact recorded at two scopes is never silently duplicated and - contradictory (SCN-010 §5.7). - """ - for name, rel in self._s.relationships.items(): - upstream = rel.proxy_upstream - if upstream is None: - continue - label = f"Relationship '{name}'" - target_node_name = self._check_proxy_upstream_node_ref( - upstream.upstream_node_ref, - label, - context="proxy_upstream", - field_name="upstream_node_ref", - ) - self._check_proxy_upstream_service_ref( - upstream.upstream_service_ref, - upstream_node_ref=target_node_name or "", - relationship_target=rel.target, - label=label, - context="proxy_upstream", - field_name="upstream_service_ref", - ) - route = self._check_proxy_upstream_route_ref(upstream.route_ref, rel.source, label) - if route is not None: - self._check_proxy_upstream_agreement(upstream, route, label, relationship_target=rel.target) - - def _check_proxy_upstream_node_ref( - self, - node_ref: str, - label: str, - *, - context: str, - field_name: str, - ) -> str | None: - if not node_ref or self._is_unresolved_var(node_ref): - return None - if node_ref not in self._s.nodes: - self._err(f"{label} {context} {field_name} '{node_ref}' does not resolve to a defined node") - return None - return node_ref - - def _check_proxy_upstream_service_ref( - self, - service_ref: str, - *, - upstream_node_ref: str, - relationship_target: str, - label: str, - context: str, - field_name: str, - ) -> None: - if not service_ref or self._is_unresolved_var(service_ref): - return - resolved = self._resolve_upstream_service_ref( - service_ref, - upstream_node_ref=upstream_node_ref, - relationship_target=relationship_target, - ) - if resolved is None: - self._err( - f"{label} {context} {field_name} '{service_ref}' cannot be resolved without a concrete upstream node" - ) - return - node_name, service_name = resolved - node = self._proxy_upstream_node( - node_name, - service_ref, - upstream_node_ref=upstream_node_ref, - relationship_target=relationship_target, - label=label, - context=context, - field_name=field_name, - ) - if node is None: - return - if service_name not in self._node_service_names(node): - self._err( - f"{label} {context} {field_name} '{service_ref}' does not resolve to a service on node '{node_name}'" - ) - - def _proxy_upstream_node( - self, - node_name: str, - service_ref: str, - *, - upstream_node_ref: str, - relationship_target: str, - label: str, - context: str, - field_name: str, - ) -> object | None: - expected_node_name = upstream_node_ref or self._node_name_from_relationship_target(relationship_target) - if expected_node_name and node_name != expected_node_name: - self._err( - f"{label} {context} {field_name} '{service_ref}' must reference a service " - f"on upstream node '{expected_node_name}'" - ) - return None - node = self._s.nodes.get(node_name) - if node is None: - self._err(f"{label} {context} upstream service node '{node_name}' does not resolve to a defined node") - return None - return node - - def _resolve_upstream_service_ref( - self, - service_ref: str, - *, - upstream_node_ref: str, - relationship_target: str, - ) -> tuple[str, str] | None: - split = self._split_node_service_ref(service_ref) - if split is not None: - return split - node_name = "" - if upstream_node_ref and not self._is_unresolved_var(upstream_node_ref): - node_name = upstream_node_ref - else: - target_node_name = self._node_name_from_relationship_target(relationship_target) - if target_node_name is not None: - node_name = target_node_name - if not node_name: - return None - return node_name, service_ref - - def _node_name_from_relationship_target(self, target: object) -> str | None: - if not isinstance(target, str) or self._is_unresolved_var(target): - return None - if target in self._s.nodes: - return target - return self._node_name_from_qualified_target(target) - - def _node_name_from_qualified_target(self, target: str) -> str | None: - service_split = self._split_node_service_ref(target) - if service_split is not None: - node_name = service_split[0] - return node_name if node_name in self._s.nodes else None - if target.startswith(_NODES_PREFIX): - node_name, sep, _tail = target[len(_NODES_PREFIX) :].partition(".runtime.") - if sep and node_name in self._s.nodes: - return node_name - return None - - def _check_proxy_upstream_route_ref(self, route_ref: str, source: str, label: str) -> object | None: - if not route_ref or self._is_unresolved_var(route_ref): - return None - routes = self._source_application_routes(source) - if routes is None: - # The source does not resolve to a runtime application surface; the - # generic relationship endpoint check already reports an unresolved - # source, so the route_ref check is deferred rather than duplicated. - return None - route = routes.get(route_ref) - if route is None: - self._err( - f"{label} proxy_upstream route_ref '{route_ref}' does not resolve to an " - f"application route on source '{source}'" - ) - return route - - def _source_application_routes(self, source: str) -> dict[str, object] | None: - """Collect ``route_id``->route for every application surface on ``source``. - - ``source`` may be a qualified ``nodes..runtime.applications.`` - ref or a bare node name; either way the proxy route lives on that node's - application surface(s). - """ - application = self._resolve_application_ref(source) - if application is not None: - return {route.route_id: route for route in application.routes} - node = self._s.nodes.get(source) - runtime = getattr(node, "runtime", None) if node is not None else None - if runtime is None: - return None - routes: dict[str, object] = {} - for application in getattr(runtime, "applications", []): - for route in application.routes: - routes[route.route_id] = route - return routes - - def _check_proxy_upstream_agreement( - self, - upstream: object, - route: object, - label: str, - *, - relationship_target: str, - ) -> None: - target = getattr(route, "upstream_target", None) - if target is None: - return - self._assert_shared_field_agreement( - label, - field_label="upstream node", - relationship_value=getattr(upstream, "upstream_node_ref", ""), - route_value=getattr(target, "target_node_ref", ""), - ) - self._assert_shared_field_agreement( - label, - field_label="upstream service", - relationship_value=getattr(upstream, "upstream_service_ref", ""), - route_value=getattr(target, "target_service", ""), - upstream_node_ref=getattr(upstream, "upstream_node_ref", "") or getattr(target, "target_node_ref", ""), - relationship_target=relationship_target, - ) - self._assert_shared_bool_agreement( - label, - field_label="TLS-termination", - relationship_value=getattr(upstream, "client_tls_terminated", None), - route_value=getattr(target, "tls_terminated_here", None), - ) - - def _assert_shared_field_agreement( - self, - label: str, - *, - field_label: str, - relationship_value: str, - route_value: str, - upstream_node_ref: str = "", - relationship_target: str = "", - ) -> None: - if not relationship_value or self._is_unresolved_var(relationship_value): - return - if not route_value or self._is_unresolved_var(route_value): - return - if self._shared_field_values_agree( - field_label=field_label, - relationship_value=relationship_value, - route_value=route_value, - upstream_node_ref=upstream_node_ref, - relationship_target=relationship_target, - ): - return - else: - self._err( - f"{label} proxy_upstream {field_label} '{relationship_value}' disagrees with the " - f"route's upstream_target value '{route_value}'" - ) - - def _shared_field_values_agree( - self, - *, - field_label: str, - relationship_value: str, - route_value: str, - upstream_node_ref: str, - relationship_target: str, - ) -> bool: - if field_label != "upstream service": - return relationship_value == route_value - relationship_ref = self._resolve_upstream_service_ref( - relationship_value, - upstream_node_ref=upstream_node_ref, - relationship_target=relationship_target, - ) - route_ref = self._resolve_upstream_service_ref( - route_value, - upstream_node_ref=upstream_node_ref, - relationship_target=relationship_target, - ) - if relationship_ref is None or route_ref is None: - return relationship_value == route_value - return relationship_ref == route_ref - - def _assert_shared_bool_agreement( - self, label: str, *, field_label: str, relationship_value: object, route_value: object - ) -> None: - if not isinstance(relationship_value, bool) or not isinstance(route_value, bool): - return - if relationship_value != route_value: - self._err( - f"{label} proxy_upstream {field_label} '{relationship_value}' disagrees with the " - f"route's upstream_target value '{route_value}'" - ) diff --git a/implementations/python/packages/aces_sdl/validator/_relationships_proxy.py b/implementations/python/packages/aces_sdl/validator/_relationships_proxy.py new file mode 100644 index 000000000..bb5dd34be --- /dev/null +++ b/implementations/python/packages/aces_sdl/validator/_relationships_proxy.py @@ -0,0 +1,293 @@ +"""SemanticValidator _RelationshipsProxyMixin (split from _relationships.py). + +Part of the SemanticValidator mixin composition; see __init__.py. +""" + +from ._support import _NODES_PREFIX + + +class _RelationshipsProxyMixin: + def _verify_relationship_proxy_upstreams(self) -> None: + """Validate typed ``proxy_upstream`` blocks on relationship edges. + + ``route_ref`` must resolve to an application route (by ``route_id``) on + the relationship's ``source`` proxy; ``upstream_node_ref`` / + ``upstream_service_ref`` (when concrete) must resolve. AGREEMENT GUARD: + when the referenced route ALSO carries an ``upstream_target``, the shared + facts (target node, target service, and the TLS-termination boolean) MUST + agree between ``route.upstream_target`` and the ``RelationshipProxyUpstream`` + so the same fact recorded at two scopes is never silently duplicated and + contradictory (SCN-010 §5.7). + """ + for name, rel in self._s.relationships.items(): + upstream = rel.proxy_upstream + if upstream is None: + continue + label = f"Relationship '{name}'" + target_node_name = self._check_proxy_upstream_node_ref( + upstream.upstream_node_ref, + label, + context="proxy_upstream", + field_name="upstream_node_ref", + ) + self._check_proxy_upstream_service_ref( + upstream.upstream_service_ref, + upstream_node_ref=target_node_name or "", + relationship_target=rel.target, + label=label, + context="proxy_upstream", + field_name="upstream_service_ref", + ) + route = self._check_proxy_upstream_route_ref(upstream.route_ref, rel.source, label) + if route is not None: + self._check_proxy_upstream_agreement(upstream, route, label, relationship_target=rel.target) + + def _check_proxy_upstream_node_ref( + self, + node_ref: str, + label: str, + *, + context: str, + field_name: str, + ) -> str | None: + if not node_ref or self._is_unresolved_var(node_ref): + return None + if node_ref not in self._s.nodes: + self._err(f"{label} {context} {field_name} '{node_ref}' does not resolve to a defined node") + return None + return node_ref + + def _check_proxy_upstream_service_ref( + self, + service_ref: str, + *, + upstream_node_ref: str, + relationship_target: str, + label: str, + context: str, + field_name: str, + ) -> None: + if not service_ref or self._is_unresolved_var(service_ref): + return + resolved = self._resolve_upstream_service_ref( + service_ref, + upstream_node_ref=upstream_node_ref, + relationship_target=relationship_target, + ) + if resolved is None: + self._err( + f"{label} {context} {field_name} '{service_ref}' cannot be resolved without a concrete upstream node" + ) + return + node_name, service_name = resolved + node = self._proxy_upstream_node( + node_name, + service_ref, + upstream_node_ref=upstream_node_ref, + relationship_target=relationship_target, + label=label, + context=context, + field_name=field_name, + ) + if node is None: + return + if service_name not in self._node_service_names(node): + self._err( + f"{label} {context} {field_name} '{service_ref}' does not resolve to a service on node '{node_name}'" + ) + + def _proxy_upstream_node( + self, + node_name: str, + service_ref: str, + *, + upstream_node_ref: str, + relationship_target: str, + label: str, + context: str, + field_name: str, + ) -> object | None: + expected_node_name = upstream_node_ref or self._node_name_from_relationship_target(relationship_target) + if expected_node_name and node_name != expected_node_name: + self._err( + f"{label} {context} {field_name} '{service_ref}' must reference a service " + f"on upstream node '{expected_node_name}'" + ) + return None + node = self._s.nodes.get(node_name) + if node is None: + self._err(f"{label} {context} upstream service node '{node_name}' does not resolve to a defined node") + return None + return node + + def _resolve_upstream_service_ref( + self, + service_ref: str, + *, + upstream_node_ref: str, + relationship_target: str, + ) -> tuple[str, str] | None: + split = self._split_node_service_ref(service_ref) + if split is not None: + return split + node_name = "" + if upstream_node_ref and not self._is_unresolved_var(upstream_node_ref): + node_name = upstream_node_ref + else: + target_node_name = self._node_name_from_relationship_target(relationship_target) + if target_node_name is not None: + node_name = target_node_name + if not node_name: + return None + return node_name, service_ref + + def _node_name_from_relationship_target(self, target: object) -> str | None: + if not isinstance(target, str) or self._is_unresolved_var(target): + return None + if target in self._s.nodes: + return target + return self._node_name_from_qualified_target(target) + + def _node_name_from_qualified_target(self, target: str) -> str | None: + service_split = self._split_node_service_ref(target) + if service_split is not None: + node_name = service_split[0] + return node_name if node_name in self._s.nodes else None + if target.startswith(_NODES_PREFIX): + node_name, sep, _tail = target[len(_NODES_PREFIX) :].partition(".runtime.") + if sep and node_name in self._s.nodes: + return node_name + return None + + def _check_proxy_upstream_route_ref(self, route_ref: str, source: str, label: str) -> object | None: + if not route_ref or self._is_unresolved_var(route_ref): + return None + routes = self._source_application_routes(source) + if routes is None: + # The source does not resolve to a runtime application surface; the + # generic relationship endpoint check already reports an unresolved + # source, so the route_ref check is deferred rather than duplicated. + return None + route = routes.get(route_ref) + if route is None: + self._err( + f"{label} proxy_upstream route_ref '{route_ref}' does not resolve to an " + f"application route on source '{source}'" + ) + return route + + def _source_application_routes(self, source: str) -> dict[str, object] | None: + """Collect ``route_id``->route for every application surface on ``source``. + + ``source`` may be a qualified ``nodes..runtime.applications.`` + ref or a bare node name; either way the proxy route lives on that node's + application surface(s). + """ + application = self._resolve_application_ref(source) + if application is not None: + return {route.route_id: route for route in application.routes} + node = self._s.nodes.get(source) + runtime = getattr(node, "runtime", None) if node is not None else None + if runtime is None: + return None + routes: dict[str, object] = {} + for application in getattr(runtime, "applications", []): + for route in application.routes: + routes[route.route_id] = route + return routes + + def _check_proxy_upstream_agreement( + self, + upstream: object, + route: object, + label: str, + *, + relationship_target: str, + ) -> None: + target = getattr(route, "upstream_target", None) + if target is None: + return + self._assert_shared_field_agreement( + label, + field_label="upstream node", + relationship_value=getattr(upstream, "upstream_node_ref", ""), + route_value=getattr(target, "target_node_ref", ""), + ) + self._assert_shared_field_agreement( + label, + field_label="upstream service", + relationship_value=getattr(upstream, "upstream_service_ref", ""), + route_value=getattr(target, "target_service", ""), + upstream_node_ref=getattr(upstream, "upstream_node_ref", "") or getattr(target, "target_node_ref", ""), + relationship_target=relationship_target, + ) + self._assert_shared_bool_agreement( + label, + field_label="TLS-termination", + relationship_value=getattr(upstream, "client_tls_terminated", None), + route_value=getattr(target, "tls_terminated_here", None), + ) + + def _assert_shared_field_agreement( + self, + label: str, + *, + field_label: str, + relationship_value: str, + route_value: str, + upstream_node_ref: str = "", + relationship_target: str = "", + ) -> None: + if not relationship_value or self._is_unresolved_var(relationship_value): + return + if not route_value or self._is_unresolved_var(route_value): + return + if self._shared_field_values_agree( + field_label=field_label, + relationship_value=relationship_value, + route_value=route_value, + upstream_node_ref=upstream_node_ref, + relationship_target=relationship_target, + ): + return + else: + self._err( + f"{label} proxy_upstream {field_label} '{relationship_value}' disagrees with the " + f"route's upstream_target value '{route_value}'" + ) + + def _shared_field_values_agree( + self, + *, + field_label: str, + relationship_value: str, + route_value: str, + upstream_node_ref: str, + relationship_target: str, + ) -> bool: + if field_label != "upstream service": + return relationship_value == route_value + relationship_ref = self._resolve_upstream_service_ref( + relationship_value, + upstream_node_ref=upstream_node_ref, + relationship_target=relationship_target, + ) + route_ref = self._resolve_upstream_service_ref( + route_value, + upstream_node_ref=upstream_node_ref, + relationship_target=relationship_target, + ) + if relationship_ref is None or route_ref is None: + return relationship_value == route_value + return relationship_ref == route_ref + + def _assert_shared_bool_agreement( + self, label: str, *, field_label: str, relationship_value: object, route_value: object + ) -> None: + if not isinstance(relationship_value, bool) or not isinstance(route_value, bool): + return + if relationship_value != route_value: + self._err( + f"{label} proxy_upstream {field_label} '{relationship_value}' disagrees with the " + f"route's upstream_target value '{route_value}'" + ) diff --git a/implementations/python/packages/aces_sdl/validator/_runtime_orchestration.py b/implementations/python/packages/aces_sdl/validator/_runtime_orchestration.py new file mode 100644 index 000000000..14022a002 --- /dev/null +++ b/implementations/python/packages/aces_sdl/validator/_runtime_orchestration.py @@ -0,0 +1,97 @@ +"""SemanticValidator _RuntimeOrchestrationMixin (split from _runtime_platform.py). + +Part of the SemanticValidator mixin composition; see __init__.py. +""" + +from .._base import is_variable_ref +from ..runtime_mounts import RuntimeControlInterfaceAccess, RuntimeControlInterfaceKind +from ..runtime_orchestration import RuntimeOrchestrationPrivilegeClass + + +class _RuntimeOrchestrationMixin: + def _verify_runtime_orchestration_authorities(self) -> None: + """Validate observed container-spawn orchestration-authority inventories. + + Each authority's ``control_interface_ref``, when present and concrete, + must resolve to a :class:`RuntimeControlInterface` declared in the same + node's ``runtime.local_control_interfaces`` (by ``control_interface_id``). + For a ``host_root_equivalent`` privilege class, the referenced control + interface must additionally be a read-write docker socket (a read-write + unix socket whose path is a ``docker.sock``), making the host-root + privilege-escalation fact resolvable at scenario scope. The + model-local ``require_profile_for_privilege_class`` guard has already + rejected a host-root-equivalent authority that carries no concrete + ``control_interface_ref``. + """ + for node_name, node in self._s.nodes.items(): + runtime = getattr(node, "runtime", None) + if runtime is None or not runtime.orchestration_authorities: + continue + interfaces_by_id = { + interface.control_interface_id: interface + for interface in getattr(runtime, "local_control_interfaces", []) + if interface.control_interface_id + } + for authority in runtime.orchestration_authorities: + self._verify_orchestration_authority( + node_name=node_name, + authority=authority, + interfaces_by_id=interfaces_by_id, + ) + + def _verify_orchestration_authority( + self, + *, + node_name: str, + authority: object, + interfaces_by_id: dict[str, object], + ) -> None: + owner_label = f"Node '{node_name}' runtime orchestration authority '{authority.orchestration_authority_id}'" + ref = getattr(authority, "control_interface_ref", "") + if not ref or self._is_unresolved_var(ref): + return + interface = interfaces_by_id.get(ref) + if interface is None: + self._err( + f"{owner_label} control_interface_ref '{ref}' does not resolve to a " + f"control interface in the same node's runtime.local_control_interfaces" + ) + return + privilege = getattr(authority, "privilege_class", None) + if ( + isinstance(privilege, RuntimeOrchestrationPrivilegeClass) + and privilege is RuntimeOrchestrationPrivilegeClass.HOST_ROOT_EQUIVALENT + ): + self._verify_host_root_control_interface(owner_label=owner_label, ref=ref, interface=interface) + + @staticmethod + def _control_interface_is_docker_socket(interface: object) -> bool: + """Return whether a control interface is a read-write docker unix socket.""" + access = getattr(interface, "access", None) + kind = getattr(interface, "kind", None) + path = getattr(interface, "path", "") or "" + is_read_write = access is RuntimeControlInterfaceAccess.READ_WRITE + is_unix_socket = kind is RuntimeControlInterfaceKind.UNIX_SOCKET + is_docker_sock = isinstance(path, str) and path.endswith("docker.sock") + return is_read_write and is_unix_socket and is_docker_sock + + def _verify_host_root_control_interface( + self, + *, + owner_label: str, + ref: str, + interface: object, + ) -> None: + # ``${var}`` placeholders on the interface's access/kind/path are + # permissive: a deferred discriminator cannot be proven non-conformant. + access = getattr(interface, "access", None) + kind = getattr(interface, "kind", None) + path = getattr(interface, "path", "") or "" + if is_variable_ref(access) or is_variable_ref(kind) or is_variable_ref(path): + return + if not self._control_interface_is_docker_socket(interface): + self._err( + f"{owner_label} privilege_class 'host_root_equivalent' control_interface_ref '{ref}' " + f"must resolve to a read-write docker socket " + f"(access 'read_write', kind 'unix_socket', path ending in 'docker.sock')" + ) diff --git a/implementations/python/packages/aces_sdl/validator/_runtime_platform.py b/implementations/python/packages/aces_sdl/validator/_runtime_platform.py index 797e0fdb1..94b195d58 100644 --- a/implementations/python/packages/aces_sdl/validator/_runtime_platform.py +++ b/implementations/python/packages/aces_sdl/validator/_runtime_platform.py @@ -5,10 +5,6 @@ from collections import defaultdict -from .._base import is_variable_ref -from ..runtime_mounts import RuntimeControlInterfaceAccess, RuntimeControlInterfaceKind -from ..runtime_orchestration import RuntimeOrchestrationPrivilegeClass - class _RuntimePlatformMixin: def _verify_runtime_security_monitoring_managers(self) -> None: @@ -465,90 +461,3 @@ def _verify_forwarding_ship_target( f"{owner_label} target_service_ref '{service_ref}' does not resolve to a service " f"on node '{resolved_node_name}'" ) - - def _verify_runtime_orchestration_authorities(self) -> None: - """Validate observed container-spawn orchestration-authority inventories. - - Each authority's ``control_interface_ref``, when present and concrete, - must resolve to a :class:`RuntimeControlInterface` declared in the same - node's ``runtime.local_control_interfaces`` (by ``control_interface_id``). - For a ``host_root_equivalent`` privilege class, the referenced control - interface must additionally be a read-write docker socket (a read-write - unix socket whose path is a ``docker.sock``), making the host-root - privilege-escalation fact resolvable at scenario scope. The - model-local ``require_profile_for_privilege_class`` guard has already - rejected a host-root-equivalent authority that carries no concrete - ``control_interface_ref``. - """ - for node_name, node in self._s.nodes.items(): - runtime = getattr(node, "runtime", None) - if runtime is None or not runtime.orchestration_authorities: - continue - interfaces_by_id = { - interface.control_interface_id: interface - for interface in getattr(runtime, "local_control_interfaces", []) - if interface.control_interface_id - } - for authority in runtime.orchestration_authorities: - self._verify_orchestration_authority( - node_name=node_name, - authority=authority, - interfaces_by_id=interfaces_by_id, - ) - - def _verify_orchestration_authority( - self, - *, - node_name: str, - authority: object, - interfaces_by_id: dict[str, object], - ) -> None: - owner_label = f"Node '{node_name}' runtime orchestration authority '{authority.orchestration_authority_id}'" - ref = getattr(authority, "control_interface_ref", "") - if not ref or self._is_unresolved_var(ref): - return - interface = interfaces_by_id.get(ref) - if interface is None: - self._err( - f"{owner_label} control_interface_ref '{ref}' does not resolve to a " - f"control interface in the same node's runtime.local_control_interfaces" - ) - return - privilege = getattr(authority, "privilege_class", None) - if ( - isinstance(privilege, RuntimeOrchestrationPrivilegeClass) - and privilege is RuntimeOrchestrationPrivilegeClass.HOST_ROOT_EQUIVALENT - ): - self._verify_host_root_control_interface(owner_label=owner_label, ref=ref, interface=interface) - - @staticmethod - def _control_interface_is_docker_socket(interface: object) -> bool: - """Return whether a control interface is a read-write docker unix socket.""" - access = getattr(interface, "access", None) - kind = getattr(interface, "kind", None) - path = getattr(interface, "path", "") or "" - is_read_write = access is RuntimeControlInterfaceAccess.READ_WRITE - is_unix_socket = kind is RuntimeControlInterfaceKind.UNIX_SOCKET - is_docker_sock = isinstance(path, str) and path.endswith("docker.sock") - return is_read_write and is_unix_socket and is_docker_sock - - def _verify_host_root_control_interface( - self, - *, - owner_label: str, - ref: str, - interface: object, - ) -> None: - # ``${var}`` placeholders on the interface's access/kind/path are - # permissive: a deferred discriminator cannot be proven non-conformant. - access = getattr(interface, "access", None) - kind = getattr(interface, "kind", None) - path = getattr(interface, "path", "") or "" - if is_variable_ref(access) or is_variable_ref(kind) or is_variable_ref(path): - return - if not self._control_interface_is_docker_socket(interface): - self._err( - f"{owner_label} privilege_class 'host_root_equivalent' control_interface_ref '{ref}' " - f"must resolve to a read-write docker socket " - f"(access 'read_write', kind 'unix_socket', path ending in 'docker.sock')" - )