diff --git a/notes/participant-role-management.md b/notes/participant-role-management.md index 69a4567b2..d37898991 100644 --- a/notes/participant-role-management.md +++ b/notes/participant-role-management.md @@ -11,6 +11,9 @@ relevant_packages: - vultron/core/models/participant.py - vultron/wire/as2/vocab/objects/case_participant.py - vultron/core/use_cases/query/action_rules.py + - vultron/core/predicates/participants.py + - vultron/core/behaviors/case/nodes/on_behalf_guards.py + - vultron/core/use_cases/triggers/case/add_on_behalf_status.py - test/core/models/test_participant.py --- @@ -268,3 +271,24 @@ exist and they are not interchangeable: Evaluate `RM.CLOSED` terminal rules *before* the `current == new` no-op check. Ordering them the other way silently permits a transition out of a terminal state whenever the target happens to equal the current state. + +## Assertion Authority and On-Behalf Exceptions (PRM-06) + +Participant status is self-declaratory by default (PRM-06-001, ADR-0084). Two +narrow on-behalf exceptions exist: + +- **v→V** (`CS_vf.Vf`): a Case Manager or Case Owner MAY assert vendor awareness + on behalf of a notified-but-not-joined vendor (PRM-06-003). +- **d→D** (`CS_d.D`): the same asserting actors MAY assert deployer deployment + under externally-evidenced exceptional circumstances (PRM-06-004). +- **f→F** (`CS_vf.VF`) is always Vendor-only and cannot be asserted on behalf + of another actor (PRM-06-005). + +The **Vendor-implies-V invariant** (PRM-06-002): a participant holding +`CVDRole.VENDOR` cannot assert `CS_vf.vf` (vendor-unaware) — a vendor that +has joined a case is by definition aware of it. Enforced by +`vendor_vf_invariant_ok` in `vultron/core/predicates/participants.py`. + +Implementation: `SvcAddOnBehalfStatusUseCase` in +`vultron/core/use_cases/triggers/case/add_on_behalf_status.py`; BT guards +in `vultron/core/behaviors/case/nodes/on_behalf_guards.py`. diff --git a/specs/participant-role-management.yaml b/specs/participant-role-management.yaml index 7c82e0a2e..62b7848eb 100644 --- a/specs/participant-role-management.yaml +++ b/specs/participant-role-management.yaml @@ -271,3 +271,101 @@ groups: tags: - testing - tooling +- id: PRM-06 + title: Assertion Authority and On-Behalf Exceptions + specs: + - id: PRM-06-001 + priority: MUST + kind: protocol + statement: >- + Participant status is self-declaratory by default: each participant + asserts its own RM and VFD state without external approval. This is why + they are *participant* status items (ADR-0084). + rationale: >- + The self-declaratory model keeps state authority with the role holder. + Deviations must be narrow, externally evidenced, and explicitly named + (ADR-0084). + adr: + - ADR-0084 + tags: + - protocol + lint_suppress: + - missing_story_reference + - id: PRM-06-002 + priority: MUST_NOT + kind: protocol + statement: >- + A participant holding ``CVDRole.VENDOR`` MUST NOT assert a VF state of + ``CS_vf.vf`` (vendor-unaware). Valid Vendor VF states are + ``{CS_vf.Vf, CS_vf.VF}`` (ADR-0084). + rationale: >- + A vendor that has joined a case is by definition aware of it. Allowing + a vendor to assert the vendor-unaware state (``vf``) would create an + impossible protocol record. + verification: >- + ``vendor_vf_invariant_ok`` in ``vultron/core/predicates/participants.py`` + enforces this predicate; it is called from both + ``ValidateTriggerTransitionsNode._check_vf_role`` (trigger path) and + ``CreateParticipantStatusNode._check_vf_precondition`` (BT action path). + adr: + - ADR-0084 + tags: + - protocol + - id: PRM-06-003 + priority: MAY + kind: protocol + statement: >- + A Case Manager or Case Owner MAY assert ``v→V`` (``CS_vf.Vf``) on behalf + of a Vendor-role holder when the notification or invite is evidenced (any + vendor acknowledgement — including ``Read(Invite(Case))`` — suffices). + Scoped to a vendor notified or invited but not yet — or never — a + participant (ADR-0084). + rationale: >- + Closes the vendor-awareness gap (CONCERN-2087): a vendor can be marked + informed even if it never joins the case, so case history is accurate. + verification: >- + ``SvcAddOnBehalfStatusUseCase`` implements this path; the asserting + actor's CASE_MANAGER or CASE_OWNER role is verified by + ``CheckOnBehalfAuthorizedNode`` before any write occurs. + adr: + - ADR-0084 + tags: + - protocol + - id: PRM-06-004 + priority: MAY + kind: protocol + statement: >- + A Case Manager or Case Owner MAY assert ``d→D`` (``CS_d.D``) on behalf + of a Deployer-role holder under the same externally-evidenced pattern + (a MAY, expected to be rare; deployment is normally self-reported by the + Deployer) (ADR-0084). + rationale: >- + Exceptional circumstances (e.g. the deployer no longer active) require + a narrow exception path; the on-behalf restriction keeps it from becoming + a general proxy capability. + adr: + - ADR-0084 + tags: + - protocol + - id: PRM-06-005 + priority: MUST_NOT + kind: protocol + statement: >- + ``f→F`` (fix ready, ``CS_vf.VF``) MUST NOT be asserted by any actor + other than the Vendor-role holder. Fix readiness is not externally + knowable; no on-behalf assertion is ever permitted for this transition + (ADR-0084). + rationale: >- + Allowing a non-vendor to assert fix-readiness would let the Case Actor + fabricate protocol-visible fix state, undermining the meaning of the F + event. + verification: >- + ``AddOnBehalfStatusTriggerRequest.vf_state_not_fix_ready`` rejects + ``CS_vf.VF`` at the request boundary before any BT runs. + ``CreateParticipantStatusNode._check_vf_precondition`` and + ``ValidateTriggerTransitionsNode._check_vf_role`` both enforce the + VENDOR-only rule on the self-report path. + adr: + - ADR-0084 + tags: + - protocol diff --git a/test/core/predicates/test_participants.py b/test/core/predicates/test_participants.py index c348752df..2aea7f7d4 100644 --- a/test/core/predicates/test_participants.py +++ b/test/core/predicates/test_participants.py @@ -21,7 +21,11 @@ ) from vultron.core.models.dimensions import RmDimension from vultron.core.models.participant_status import ParticipantStatus -from vultron.core.predicates.participants import all_participants_rm_closed +from vultron.core.predicates.participants import ( + all_participants_rm_closed, + vendor_vf_invariant_ok, +) +from vultron.core.states.cs import CS_vf from vultron.core.states.rm import RM from vultron.enums.roles import CVDRole @@ -137,3 +141,36 @@ def test_mixed_roles_including_case_manager_skipped(self): roles=[CVDRole.COORDINATOR, CVDRole.CASE_MANAGER], ) assert all_participants_rm_closed([p]) is True + + +class TestVendorVfInvariantOk: + """vendor_vf_invariant_ok: VENDOR participant cannot hold CS_vf.vf (ADR-0084, PRM-06-002).""" + + def test_none_vf_state_always_ok(self): + assert vendor_vf_invariant_ok([CVDRole.VENDOR], None) is True + + def test_vendor_with_vf_fails(self): + assert vendor_vf_invariant_ok([CVDRole.VENDOR], CS_vf.vf) is False + + def test_vendor_with_Vf_ok(self): + assert vendor_vf_invariant_ok([CVDRole.VENDOR], CS_vf.Vf) is True + + def test_vendor_with_VF_ok(self): + assert vendor_vf_invariant_ok([CVDRole.VENDOR], CS_vf.VF) is True + + def test_non_vendor_with_vf_ok(self): + assert vendor_vf_invariant_ok([CVDRole.COORDINATOR], CS_vf.vf) is True + + def test_non_vendor_with_Vf_ok(self): + assert vendor_vf_invariant_ok([CVDRole.COORDINATOR], CS_vf.Vf) is True + + def test_empty_roles_with_vf_ok(self): + assert vendor_vf_invariant_ok([], CS_vf.vf) is True + + def test_vendor_plus_coordinator_with_vf_fails(self): + assert ( + vendor_vf_invariant_ok( + [CVDRole.VENDOR, CVDRole.COORDINATOR], CS_vf.vf + ) + is False + ) diff --git a/test/core/use_cases/triggers/case/test_add_on_behalf_status.py b/test/core/use_cases/triggers/case/test_add_on_behalf_status.py new file mode 100644 index 000000000..a0b7d7604 --- /dev/null +++ b/test/core/use_cases/triggers/case/test_add_on_behalf_status.py @@ -0,0 +1,281 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""Tests for SvcAddOnBehalfStatusUseCase. + +Covers: +- AC-1: Case Manager asserts v→V for a notified-not-joined vendor. +- Blocked when asserting actor lacks CM/CO role. +- AC-5 (request layer): CS_vf.VF is rejected at the request boundary. +- AC-4: Vendor-implies-V invariant blocks a joined vendor from asserting vf. +""" + +import pytest + +from vultron.adapters.driven.datalayer_sqlite import ( + SqliteDataLayer, + reset_datalayer, +) +from vultron.adapters.driven.trigger_activity_adapter import ( + TriggerActivityAdapter, +) +from vultron.core.models.case_participant import CaseParticipant +from vultron.core.states.cs import CS_d, CS_vf +from vultron.core.use_cases.triggers.case import ( + AddOnBehalfStatusTriggerRequest, + AddParticipantStatusTriggerRequest, + SvcAddOnBehalfStatusUseCase, + SvcAddParticipantStatusUseCase, +) +from vultron.enums.roles import CVDRole +from vultron.errors import VultronValidationError +from vultron.wire.as2.vocab.base.objects.actors import as_Service +from vultron.wire.as2.vocab.objects.case_participant import as_CaseParticipant +from vultron.wire.as2.vocab.objects.vulnerability_case import ( + as_VulnerabilityCase, +) + + +def _make_actor(name: str) -> as_Service: + return as_Service(name=name, url=f"https://example.org/{name.lower()}") + + +def _make_actor_dl(name: str) -> tuple[as_Service, SqliteDataLayer]: + actor = _make_actor(name) + dl = SqliteDataLayer("sqlite:///:memory:", actor_id=actor.id_) + dl.clear_all() + dl.create(actor) + return actor, dl + + +def _to_ids(activity) -> list[str]: + to = getattr(activity, "to", None) + if isinstance(to, list): + return [ + item if isinstance(item, str) else getattr(item, "id_", str(item)) + for item in to + ] + if isinstance(to, str): + return [to] + return [] + + +def _make_base_case( + dl: SqliteDataLayer, + case_manager_actor_id: str, +) -> as_VulnerabilityCase: + """Return a case with one Case Manager participant; no vendor yet.""" + case = as_VulnerabilityCase(name="Test Case") + cm_participant = as_CaseParticipant( + attributed_to=case_manager_actor_id, + context=case.id_, + case_roles=[CVDRole.CASE_MANAGER], + ) + case.actor_participant_index[case_manager_actor_id] = cm_participant.id_ + case.case_participants.append(cm_participant.id_) + dl.create(case) + dl.create(cm_participant) + return case + + +class TestAddOnBehalfStatusVtoV: + """AC-1: Case Manager asserts v→V for a notified-but-not-joined vendor.""" + + @pytest.fixture(autouse=True) + def setup(self): + self.cm_actor, self.dl = _make_actor_dl("CaseManager") + self.vendor_actor = _make_actor("Vendor Co") + self.case = _make_base_case(self.dl, self.cm_actor.id_) + yield + self.dl.clear_all() + reset_datalayer(self.cm_actor.id_) + + def test_creates_participant_and_status_for_new_vendor(self): + request = AddOnBehalfStatusTriggerRequest( + actor_id=self.cm_actor.id_, + case_id=self.case.id_, + target_actor_id=self.vendor_actor.id_, + vf_state=CS_vf.Vf, + ) + result = SvcAddOnBehalfStatusUseCase( + self.dl, request, trigger_activity=TriggerActivityAdapter(self.dl) + ).execute() + + assert result.get("status_id") is not None + + # Vendor now has a CaseParticipant in the case + updated_case = self.dl.read_case(self.case.id_) + assert updated_case is not None + assert self.vendor_actor.id_ in updated_case.actor_participant_index + + # Vendor's participant has a status with VF=Vf + participant_id = updated_case.actor_participant_index[ + self.vendor_actor.id_ + ] + participant = self.dl.read(participant_id) + assert isinstance(participant, CaseParticipant) + assert participant.participant_statuses + last_status = participant.participant_statuses[-1] + assert last_status.vf is not None + assert last_status.vf.state == CS_vf.Vf + + def test_queues_outbox_activity_addressed_to_case_manager(self): + request = AddOnBehalfStatusTriggerRequest( + actor_id=self.cm_actor.id_, + case_id=self.case.id_, + target_actor_id=self.vendor_actor.id_, + vf_state=CS_vf.Vf, + ) + before = set(self.dl.outbox_list()) + SvcAddOnBehalfStatusUseCase( + self.dl, request, trigger_activity=TriggerActivityAdapter(self.dl) + ).execute() + + after = set(self.dl.outbox_list()) + new_ids = after - before + assert new_ids, "on-behalf assertion must queue an outbox activity" + activity_id = next(iter(new_ids)) + activity = self.dl.read(activity_id) + assert activity is not None + to_ids = _to_ids(activity) + assert ( + self.cm_actor.id_ in to_ids + ), f"PCR-08-001: activity must address the Case Manager; to={to_ids!r}" + + def test_blocked_when_asserting_actor_not_cm_or_co(self): + """Non-CM/CO actor cannot make an on-behalf assertion (PRM-06-003).""" + coordinator_actor = _make_actor("Coordinator") + # Register coordinator in the same DL so resolve_actor succeeds + self.dl.create(coordinator_actor) + coord_participant = as_CaseParticipant( + attributed_to=coordinator_actor.id_, + context=self.case.id_, + case_roles=[CVDRole.COORDINATOR], + ) + self.case.actor_participant_index[coordinator_actor.id_] = ( + coord_participant.id_ + ) + self.case.case_participants.append(coord_participant.id_) + self.dl.save(self.case) + self.dl.create(coord_participant) + + request = AddOnBehalfStatusTriggerRequest( + actor_id=coordinator_actor.id_, + case_id=self.case.id_, + target_actor_id=self.vendor_actor.id_, + vf_state=CS_vf.Vf, + ) + with pytest.raises(VultronValidationError): + SvcAddOnBehalfStatusUseCase( + self.dl, + request, + trigger_activity=TriggerActivityAdapter(self.dl), + ).execute() + + +class TestAddOnBehalfRequestValidation: + """AC-3 / PRM-06-005: CS_vf.VF (f→F) rejected at request boundary.""" + + def test_vf_state_VF_raises_at_request_construction(self): + with pytest.raises(ValueError, match="f→F"): + AddOnBehalfStatusTriggerRequest( + actor_id="https://example.org/cm", + case_id="https://example.org/case", + target_actor_id="https://example.org/vendor", + vf_state=CS_vf.VF, + ) + + def test_vf_state_Vf_accepted(self): + req = AddOnBehalfStatusTriggerRequest( + actor_id="https://example.org/cm", + case_id="https://example.org/case", + target_actor_id="https://example.org/vendor", + vf_state=CS_vf.Vf, + ) + assert req.vf_state == CS_vf.Vf + + def test_vf_state_none_accepted_when_d_state_provided(self): + req = AddOnBehalfStatusTriggerRequest( + actor_id="https://example.org/cm", + case_id="https://example.org/case", + target_actor_id="https://example.org/deployer", + d_state=CS_d.D, + ) + assert req.vf_state is None + assert req.d_state == CS_d.D + + def test_all_none_raises(self): + with pytest.raises(ValueError, match="at least one"): + AddOnBehalfStatusTriggerRequest( + actor_id="https://example.org/cm", + case_id="https://example.org/case", + target_actor_id="https://example.org/vendor", + ) + + +class TestVendorImpliesVInvariant: + """AC-4: A joined vendor cannot self-report CS_vf.vf (PRM-06-002).""" + + @pytest.fixture(autouse=True) + def setup(self): + self.vendor_actor, self.dl = _make_actor_dl("Vendor Co") + self.cm_actor = _make_actor("Case Manager") + self.case = _make_base_case(self.dl, self.cm_actor.id_) + + # Add the vendor as a joined participant with VENDOR role at CS_vf.Vf + from vultron.wire.as2.vocab.objects.case_participant import ( + as_CaseParticipant as WireCaseParticipant, + ) + from vultron.wire.as2.vocab.objects.case_status import ( + as_ParticipantStatus as WireParticipantStatus, + ) + + vendor_participant = WireCaseParticipant( + attributed_to=self.vendor_actor.id_, + context=self.case.id_, + case_roles=[CVDRole.VENDOR], + ) + from vultron.core.states.rm import RM + + vendor_participant.participant_statuses.append( + WireParticipantStatus( + context=self.case.id_, + rm_state=RM.ACCEPTED, + vf_state=CS_vf.Vf, + ) + ) + self.case.actor_participant_index[self.vendor_actor.id_] = ( + vendor_participant.id_ + ) + self.case.case_participants.append(vendor_participant.id_) + self.dl.save(self.case) + self.dl.create(vendor_participant) + yield + self.dl.clear_all() + reset_datalayer(self.vendor_actor.id_) + + def test_vendor_cannot_self_report_vf_unaware(self): + """A joined vendor cannot assert CS_vf.vf (vendor-unaware) via self-report.""" + request = AddParticipantStatusTriggerRequest( + actor_id=self.vendor_actor.id_, + case_id=self.case.id_, + vf_state=CS_vf.vf, + ) + with pytest.raises(VultronValidationError): + SvcAddParticipantStatusUseCase( + self.dl, + request, + trigger_activity=TriggerActivityAdapter(self.dl), + ).execute() diff --git a/test/core/use_cases/triggers/case/test_add_participant_status.py b/test/core/use_cases/triggers/case/test_add_participant_status.py index c790a3c1a..b4cb50a4f 100644 --- a/test/core/use_cases/triggers/case/test_add_participant_status.py +++ b/test/core/use_cases/triggers/case/test_add_participant_status.py @@ -898,13 +898,17 @@ def test_invalid_vf_jump_blocked_at_write_node(self): assert "status_id" not in result_out def test_same_state_vf_write_allowed_at_write_node(self): - """CSB-16-001: same-state VF write (no actual transition) is allowed.""" + """CSB-16-001: same-state VF write (no actual transition) is allowed. + + Uses CS_vf.Vf because VENDOR participants cannot hold CS_vf.vf + (Vendor-implies-V, PRM-06-002, ADR-0084). + """ from py_trees.common import Status - self._seed_participant_vf_state(CS_vf.vf) + self._seed_participant_vf_state(CS_vf.Vf) bt_result, result_out = self._run_node( - rm_state=None, vf_state=CS_vf.vf, d_state=None, pxa_state=None + rm_state=None, vf_state=CS_vf.Vf, d_state=None, pxa_state=None ) assert bt_result.status == Status.SUCCESS @@ -1856,3 +1860,172 @@ def test_p_bit_without_active_embargo_returns_none(self): from vultron.core.states.em import EM assert self._check(CS_pxa.Pxa, EM.NONE) is None + + +class TestCreateParticipantStatusNodeCrossMachineOnBypassPath: + """CreateParticipantStatusNode enforces cross-machine entailments (#3100). + + Bypass callers (DevelopFixNode, DeployFixNode, etc.) reach + CreateParticipantStatusNode without going through + ValidateTriggerTransitionsNode. The write node must reject a + CSB-18-001/CSB-17-001 violation so those callers cannot persist an + impossible RM+VF or VF+D combination. + """ + + @pytest.fixture(autouse=True) + def setup(self): + import py_trees + + from vultron.adapters.driven.datalayer_sqlite import ( + SqliteDataLayer, + reset_datalayer, + ) + from vultron.adapters.driven.trigger_activity_adapter import ( + TriggerActivityAdapter, + ) + from vultron.core.behaviors.bridge import BTBridge + from vultron.enums.roles import CVDRole + from vultron.wire.as2.vocab.base.objects.actors import as_Service + from vultron.wire.as2.vocab.objects.case_participant import ( + as_CaseParticipant, + ) + from vultron.wire.as2.vocab.objects.vulnerability_case import ( + as_VulnerabilityCase, + ) + + py_trees.blackboard.Blackboard.enable_activity_stream() + py_trees.blackboard.Blackboard.storage.clear() + + self.actor = as_Service(name="Vendor Bypass") + actor_id = self.actor.id_ + reset_datalayer(actor_id) + self.dl = SqliteDataLayer("sqlite:///:memory:", actor_id=actor_id) + self.dl.clear_all() + self.dl.create(self.actor) + + self.case_actor = as_Service(name="Case Actor Bypass") + reset_datalayer(self.case_actor.id_) + self.dl.create(self.case_actor) + + self.case = as_VulnerabilityCase(name="Test Case #3100") + self.actor_participant = as_CaseParticipant( + attributed_to=actor_id, + context=self.case.id_, + case_roles=[CVDRole.VENDOR], + ) + self.case_manager_participant = as_CaseParticipant( + attributed_to=self.case_actor.id_, + context=self.case.id_, + case_roles=[CVDRole.CASE_MANAGER], + ) + self.case.actor_participant_index[actor_id] = ( + self.actor_participant.id_ + ) + self.case.actor_participant_index[self.case_actor.id_] = ( + self.case_manager_participant.id_ + ) + self.dl.create(self.case) + self.dl.create(self.actor_participant) + self.dl.create(self.case_manager_participant) + self.bridge = BTBridge( + datalayer=self.dl, + trigger_activity=TriggerActivityAdapter(self.dl), + ) + yield + try: + self.dl.clear_all() + finally: + self.dl.close() + reset_datalayer(actor_id) + reset_datalayer(self.case_actor.id_) + py_trees.blackboard.Blackboard.storage.clear() + + def _run_node(self, **kwargs): + from vultron.core.behaviors.case.nodes.participant import ( + CreateParticipantStatusNode, + ) + + result_out: dict = {} + node = CreateParticipantStatusNode( + case_id=self.case.id_, + actor_id=self.actor.id_, + result_out=result_out, + **kwargs, + ) + bt_result = self.bridge.execute_with_setup( + node, actor_id=self.actor.id_ + ) + return bt_result, result_out + + def test_vf_fix_ready_with_rm_start_rejected_by_write_node(self): + """CSB-18-001 bypass guard (#3100): write node refuses VF=VF when RM=START. + + CreateParticipantStatusNode now calls cross_machine_violations() so a + caller that bypasses ValidateTriggerTransitionsNode cannot persist a + state that the trigger guard would have refused. + """ + from py_trees.common import Status + + bt_result, result_out = self._run_node( + rm_state=None, vf_state=CS_vf.VF, d_state=None, pxa_state=None + ) + + assert bt_result.status == Status.FAILURE + assert "status_id" not in result_out + + +def test_validate_trigger_returns_failure_on_corrupt_participant_status(): + """#3103: VultronValidationError from resolve_participant_state_from_dl is caught. + + Before the fix, a participant with a non-core-shaped status let + VultronValidationError escape update(), producing a 500. After the fix the + node returns Status.FAILURE with a descriptive feedback_message. + + Uses a stubbed DataLayer so the bad RM state bypasses the SQLite adapter's + rehydration path — matching the test pattern at line 176 in this file. + """ + from py_trees.common import Status + from vultron.core.behaviors.case.nodes.participant.trigger_validation import ( + ValidateTriggerTransitionsNode, + ) + + ACTOR_ID = "https://example.org/corrupt-vendor" + CASE_ID = "https://example.org/case-3103" + PARTICIPANT_ID = "https://example.org/participant-3103" + + class _BadRmDim: + state = "not-an-rm" + + class _CorruptStatus: + rm = _BadRmDim() + vf = None + d = None + + class _CorruptParticipant: + participant_statuses = [_CorruptStatus()] + + class _StubCase: + actor_participant_index = {ACTOR_ID: PARTICIPANT_ID} + case_participants: list = [] + + class _StubDL: + def read_case(self, case_id: str): + return _StubCase() + + def read(self, id_: str): + return _CorruptParticipant() + + node = ValidateTriggerTransitionsNode( + case_id=CASE_ID, + actor_id=ACTOR_ID, + rm_state=RM.RECEIVED, + vf_state=None, + d_state=None, + pxa_state=None, + ) + node.datalayer = _StubDL() # type: ignore[assignment] + + result = node.update() + + assert result == Status.FAILURE + assert "core-shaped" in node.feedback_message diff --git a/vultron/core/behaviors/case/add_on_behalf_status_trigger_tree.py b/vultron/core/behaviors/case/add_on_behalf_status_trigger_tree.py new file mode 100644 index 000000000..46806f387 --- /dev/null +++ b/vultron/core/behaviors/case/add_on_behalf_status_trigger_tree.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERTⓇ and CERT Coordination CenterⓇ are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""Trigger-side BT for the on-behalf v→V / d→D assertion workflow. + +The asserting actor (Case Manager or Case Owner) records vendor-awareness +or deployer-fix-deployment on behalf of a notified-but-not-yet-joined actor. +The tree runs four steps in sequence: + +1. **CheckOnBehalfAuthorizedNode** — verify the asserting actor holds + CASE_MANAGER or CASE_OWNER (ADR-0084, PRM-06-003/004). +2. **EnsureOnBehalfParticipantExistsNode** — create a minimal + ``CaseParticipant`` for the target if absent (ADR-0084). +3. **CreateParticipantStatusNode** — write the ParticipantStatus snapshot + for the target actor (BT-15-001: protocol-significant write inside BT). +4. **sender_side_bt** — resolve the Case Manager, build the outbound + ``Add(ParticipantStatus)`` activity, and queue it. +""" + +from typing import Callable + +import py_trees + +from vultron.core.behaviors.case.nodes.participant import ( + CreateParticipantStatusNode, +) +from vultron.core.behaviors.case.nodes.on_behalf_guards import ( + CheckOnBehalfAuthorizedNode, + EnsureOnBehalfParticipantExistsNode, +) +from vultron.core.behaviors.sender.send_tree import sender_side_bt +from vultron.core.states.cs import CS_d, CS_vf +from vultron.enums.roles import CVDRole + + +def add_on_behalf_status_trigger_bt( + case_id: str, + asserting_actor_id: str, + target_actor_id: str, + required_roles: list[CVDRole], + vf_state: "CS_vf | None", + d_state: "CS_d | None", + result_out: dict, + activity_builder: Callable[[str], list[str]], +) -> py_trees.behaviour.Behaviour: + """Return the trigger-side BT for the on-behalf status assertion workflow. + + Args: + case_id: ID of the VulnerabilityCase. + asserting_actor_id: Actor making the assertion (must hold CASE_MANAGER + or CASE_OWNER). + target_actor_id: Actor whose awareness/deployment is being recorded. + required_roles: Roles to assign when creating a new participant; + ``[CVDRole.VENDOR]`` for v→V, ``[CVDRole.DEPLOYER]`` for d→D, + ``[CVDRole.VENDOR, CVDRole.DEPLOYER]`` when both are requested. + vf_state: ``CS_vf.Vf`` for v→V, or ``None``. + d_state: ``CS_d.D`` for d→D, or ``None``. + result_out: Mutable dict populated by ``CreateParticipantStatusNode`` + with ``'status_id'`` and ``'participant_id'``. + activity_builder: ``(case_manager_id: str) -> list[str]`` — called by + ``sender_side_bt`` after resolving the Case Manager. + + Returns: + A ``py_trees.composites.Sequence`` that gates, creates, and emits. + """ + return py_trees.composites.Sequence( + name="AddOnBehalfStatusTriggerBT", + memory=False, + children=[ + CheckOnBehalfAuthorizedNode( + case_id=case_id, + asserting_actor_id=asserting_actor_id, + ), + EnsureOnBehalfParticipantExistsNode( + case_id=case_id, + target_actor_id=target_actor_id, + required_roles=required_roles, + ), + CreateParticipantStatusNode( + case_id=case_id, + actor_id=target_actor_id, + rm_state=None, + vf_state=vf_state, + d_state=d_state, + pxa_state=None, + result_out=result_out, + ), + sender_side_bt(case_id=case_id, activity_builder=activity_builder), + ], + ) diff --git a/vultron/core/behaviors/case/nodes/__init__.py b/vultron/core/behaviors/case/nodes/__init__.py index e8904c902..25a0e0124 100644 --- a/vultron/core/behaviors/case/nodes/__init__.py +++ b/vultron/core/behaviors/case/nodes/__init__.py @@ -128,6 +128,10 @@ CheckIsCaseOwnerNode, CheckNotSoleObserverVfdNode, ) +from vultron.core.behaviors.case.nodes.on_behalf_guards import ( + CheckOnBehalfAuthorizedNode, + EnsureOnBehalfParticipantExistsNode, +) from vultron.core.behaviors.case.nodes.update import ( ApplyCaseUpdateNode, BroadcastCaseUpdateNode, @@ -201,6 +205,9 @@ "create_case_manager_gated_tree", # vfd_role_guards (condition nodes) "CheckNotSoleObserverVfdNode", + # on_behalf_guards (ADR-0084) + "CheckOnBehalfAuthorizedNode", + "EnsureOnBehalfParticipantExistsNode", # suggest_actor (leaf nodes) "ActorAlreadyParticipantNode", "EmitAcceptActorRecommendationNode", diff --git a/vultron/core/behaviors/case/nodes/on_behalf_guards.py b/vultron/core/behaviors/case/nodes/on_behalf_guards.py new file mode 100644 index 000000000..ce32611c0 --- /dev/null +++ b/vultron/core/behaviors/case/nodes/on_behalf_guards.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERTⓇ and CERT Coordination CenterⓇ are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""On-behalf assertion guard nodes for the add-on-behalf-status trigger. + +Implements the narrow externally-evidenced on-behalf exceptions from ADR-0084: + +- :class:`CheckOnBehalfAuthorizedNode` — on-behalf assertion gate: + asserting actor MUST hold ``CVDRole.CASE_MANAGER`` or ``CVDRole.CASE_OWNER`` + (ADR-0084, PRM-06-003/004) +- :class:`EnsureOnBehalfParticipantExistsNode` — creates a minimal + ``CaseParticipant`` for the target actor when absent from the case + (ADR-0084, PRM-06-003/004) +""" + +import logging + +from py_trees.common import Status + +from vultron.core.behaviors.helpers import ( + DataLayerActionWithPorts, + DataLayerConditionWithPorts, +) +from vultron.core.behaviors.case.nodes.participant.common import ( + _create_and_attach_participant, +) +from vultron.core.behaviors.case.nodes.vfd_role_guards import ( + _resolve_actor_roles, +) +from vultron.core.models.case_participant import CaseParticipant +from vultron.enums.roles import CVDRole + +logger = logging.getLogger(__name__) + + +class CheckOnBehalfAuthorizedNode(DataLayerConditionWithPorts): + """Gate on-behalf assertions: asserting actor MUST hold CASE_MANAGER or CASE_OWNER. + + Used as the first guard in the on-behalf status trigger tree (ADR-0084, + PRM-06-003/004). Returns ``SUCCESS`` when the actor holds either + management role; ``FAILURE`` otherwise. + """ + + def __init__( + self, + case_id: str, + asserting_actor_id: str, + name: str | None = None, + ) -> None: + super().__init__(name=name or self.__class__.__name__) + self._case_id = case_id + self._asserting_actor_id = asserting_actor_id + + def update(self) -> Status: + if (f := self._require_datalayer()) is not None: + return f + assert self.datalayer is not None + + roles = _resolve_actor_roles( + self.datalayer, self._case_id, self._asserting_actor_id, self.name + ) + if roles is None: + self.feedback_message = ( + f"Could not resolve roles for actor '{self._asserting_actor_id}'" + f" in case '{self._case_id}'" + ) + return Status.FAILURE + + authorized = {CVDRole.CASE_MANAGER, CVDRole.CASE_OWNER} + if not authorized.intersection(roles): + self.feedback_message = ( + f"Actor '{self._asserting_actor_id}' does not hold" + f" CASE_MANAGER or CASE_OWNER in case '{self._case_id}'" + f" — on-behalf assertion blocked (PRM-06-003, ADR-0084)" + f" (roles={roles!r})" + ) + self.logger.warning("%s: %s", self.name, self.feedback_message) + return Status.FAILURE + + self.logger.debug( + "%s: actor '%s' is authorized for on-behalf assertion (roles=%s)", + self.name, + self._asserting_actor_id, + roles, + ) + return Status.SUCCESS + + +class EnsureOnBehalfParticipantExistsNode(DataLayerActionWithPorts): + """Ensure the target actor has a CaseParticipant; create one if absent. + + For on-behalf v→V (AC-1) and d→D (AC-2): the target (vendor or deployer) + may not yet be a case participant. This node looks up the target in + ``actor_participant_index``; if absent, creates a minimal ``CaseParticipant`` + with ``required_roles`` and attaches it to the case so that + ``CreateParticipantStatusNode`` can append a status to it. + + When both ``vf_state`` and ``d_state`` are requested on the same actor + (e.g. a combined vendor-deployer), pass both roles so the single new + ``CaseParticipant`` satisfies both the VF and D precondition checks. + + Returns ``SUCCESS`` when the participant exists or was just created. + Returns ``FAILURE`` if the case cannot be resolved. + + Per ADR-0084, PRM-06-003/004. + """ + + def __init__( + self, + case_id: str, + target_actor_id: str, + required_roles: list[CVDRole], + name: str | None = None, + ) -> None: + super().__init__(name=name or self.__class__.__name__) + self._case_id = case_id + self._target_actor_id = target_actor_id + self._required_roles = required_roles + + def update(self) -> Status: + if (f := self._require_datalayer()) is not None: + return f + assert self.datalayer is not None + dl = self.datalayer + + case = dl.read_case(self._case_id) + if case is None: + self.feedback_message = f"Case '{self._case_id}' not found" + self.logger.error("%s: %s", self.name, self.feedback_message) + return Status.FAILURE + + if self._target_actor_id in case.actor_participant_index: + self.logger.debug( + "%s: target actor '%s' already has a participant in case '%s'", + self.name, + self._target_actor_id, + self._case_id, + ) + return Status.SUCCESS + + participant = CaseParticipant( + attributed_to=self._target_actor_id, + context=self._case_id, + case_roles=self._required_roles, + ) + updated_case = _create_and_attach_participant( + dl, + participant, + self._case_id, + self._target_actor_id, + self.logger, + ) + if updated_case is None: + self.feedback_message = ( + f"Failed to create/attach participant for" + f" '{self._target_actor_id}' in case '{self._case_id}'" + ) + return Status.FAILURE + + dl.save(updated_case) + self.logger.info( + "%s: created on-behalf participant '%s' with roles %s in case '%s'", + self.name, + self._target_actor_id, + self._required_roles, + self._case_id, + ) + return Status.SUCCESS diff --git a/vultron/core/behaviors/case/nodes/participant/status.py b/vultron/core/behaviors/case/nodes/participant/status.py index c76f47ab0..0e2feca3e 100644 --- a/vultron/core/behaviors/case/nodes/participant/status.py +++ b/vultron/core/behaviors/case/nodes/participant/status.py @@ -50,11 +50,15 @@ is_valid_pxa_transition, is_valid_vf_transition, ) +from vultron.core.states.cross_machine_invariants import ( + cross_machine_violations, +) from vultron.core.states.cs_invariants import ( cs_from_dimensions, is_valid_cs_transition, ) from vultron.core.states.em import EM +from vultron.core.predicates.participants import vendor_vf_invariant_ok from vultron.core.states.rm import RM from vultron.core.predicates.roles import has_deployer_role, has_vendor_role @@ -198,12 +202,25 @@ def _build_participant_metadata( def _check_vf_precondition( self, current_vf: CS_vf | None, participant_obj: object ) -> "Status | None": - """CSB-16-001 / ADR-0075 / CSB-15-001: validate VF transition and role before writing.""" + """CSB-16-001 / ADR-0075 / CSB-15-001 / PRM-06-002: validate VF transition and role before writing.""" actor_roles = ( participant_obj.roles # type: ignore[union-attr] if isinstance(participant_obj, CaseParticipant) else [] ) + if not vendor_vf_invariant_ok(actor_roles, self._vf_state): + self.logger.warning( + "%s: Vendor-implies-V violated: actor '%s' holds VENDOR but" + " asserts %s (PRM-06-002, ADR-0084)", + self.name, + self._actor_id, + self._vf_state, + ) + self.feedback_message = ( + f"Vendor-implies-V: VENDOR cannot assert" + f" {self._vf_state!r} (PRM-06-002)" + ) + return Status.FAILURE if self._vf_state in (CS_vf.Vf, CS_vf.VF) and not has_vendor_role( actor_roles ): @@ -365,6 +382,7 @@ def _validate_transitions( current_vf: "CS_vf | None", current_d: "CS_d | None", pxa_before: CS_pxa, + eff_rm: "RM", eff_vf: "CS_vf | None", eff_d: "CS_d | None", eff_pxa: CS_pxa, @@ -377,6 +395,10 @@ def _validate_transitions( check is only run when all per-dimension checks pass — it is derived when any single-dimension violation is already present (EH-07-002). + Also enforces cross-machine entailments (CSB-18-001, CSB-17-001) so + bypass callers (DevelopFixNode, DeployFixNode, etc.) cannot persist a + state that ValidateTriggerTransitionsNode would have refused (#3100). + Returns ``Status.FAILURE`` with a joined ``feedback_message`` when any check fails; ``None`` when all checks pass. """ @@ -402,6 +424,10 @@ def _validate_transitions( ): errors.append(self.feedback_message) + if not errors: + for violation in cross_machine_violations(eff_rm, eff_vf, eff_d): + errors.append(violation.message) + if errors: self.feedback_message = "; ".join(errors) return Status.FAILURE @@ -445,6 +471,7 @@ def update(self) -> Status: pxa_before = _resolve_pxa_state(case, participant_obj) # Effective states before promotion (what the caller requested) + eff_rm = self._rm_state if self._rm_state is not None else current_rm eff_vf = self._vf_state if self._vf_state is not None else current_vf eff_d = self._d_state if self._d_state is not None else current_d eff_pxa = ( @@ -455,6 +482,7 @@ def update(self) -> Status: current_vf, current_d, pxa_before, + eff_rm, eff_vf, eff_d, eff_pxa, diff --git a/vultron/core/behaviors/case/nodes/participant/trigger_validation.py b/vultron/core/behaviors/case/nodes/participant/trigger_validation.py index b4509ab22..f3a56eb85 100644 --- a/vultron/core/behaviors/case/nodes/participant/trigger_validation.py +++ b/vultron/core/behaviors/case/nodes/participant/trigger_validation.py @@ -45,6 +45,7 @@ ) from vultron.core.behaviors.helpers import DataLayerCondition from vultron.core.models.case_participant import CaseParticipant +from vultron.errors import VultronValidationError from vultron.core.states.cross_machine_invariants import ( cross_machine_violations, ) @@ -56,6 +57,7 @@ is_valid_pxa_transition, is_valid_vf_transition, ) +from vultron.core.predicates.participants import vendor_vf_invariant_ok from vultron.core.states.rm import RM, is_valid_rm_transition from vultron.core.predicates.roles import has_vendor_role @@ -129,27 +131,82 @@ def __init__( self._pxa_state = pxa_state def _check_vf_role(self, participant_obj: object) -> "Status | None": - """Return FAILURE when the requested VF state requires VENDOR but actor lacks it. + """Return FAILURE when the VF assertion violates a role rule. - Vendor-aware VF states (Vf, VF) are VENDOR-specific per ADR-0075. - Returns None when no VF state is requested, the state is CS_vf.vf - (vendor-unaware), or the actor holds CVDRole.VENDOR. Closes #2862. + Two rules are enforced (ADR-0075, ADR-0084): + + * AC-4 / PRM-06-002 — a VENDOR-role participant cannot self-assert + ``CS_vf.vf`` (vendor-unaware): they are by definition already aware. + * ADR-0075 / CSB-15-001 — only VENDOR may assert Vf or VF; a non-vendor + participant cannot claim vendor-awareness or fix-readiness. + + Returns ``None`` (pass) when no VF state is requested. """ - if self._vf_state is None or self._vf_state == CS_vf.vf: + if self._vf_state is None: return None actor_roles = ( list(participant_obj.roles) # type: ignore[attr-defined] if isinstance(participant_obj, CaseParticipant) else [] ) - if has_vendor_role(actor_roles): + if not vendor_vf_invariant_ok(actor_roles, self._vf_state): + self.feedback_message = ( + f"Vendor-implies-V: CVDRole.VENDOR participant cannot assert" + f" {self._vf_state!r} (PRM-06-002, ADR-0084)" + ) + self.logger.info("%s: %s", self.name, self.feedback_message) + return Status.FAILURE + if self._vf_state == CS_vf.vf: + return None # non-vendor asserting vf is valid + if not has_vendor_role(actor_roles): + self.feedback_message = ( + f"CVDRole.VENDOR required for VF state" + f" {self._vf_state!r} (ADR-0075); actor roles: {actor_roles!r}" + ) + self.logger.info("%s: %s", self.name, self.feedback_message) + return Status.FAILURE + return None + + def _check_pxa_transition( + self, case: object, participant_obj: object + ) -> "Status | None": + """Validate the requested PXA transition; return FAILURE or None.""" + if self._pxa_state is None or not isinstance( + participant_obj, CaseParticipant + ): return None - self.feedback_message = ( - f"CVDRole.VENDOR required for VF state" - f" {self._vf_state!r} (ADR-0075); actor roles: {actor_roles!r}" - ) - self.logger.info("%s: %s", self.name, self.feedback_message) - return Status.FAILURE + current_pxa = _resolve_current_pxa(case, participant_obj) + if self._pxa_state != current_pxa and not is_valid_pxa_transition( + current_pxa, self._pxa_state + ): + self.feedback_message = ( + f"Invalid PXA transition" + f" {current_pxa!r} → {self._pxa_state!r}" + ) + self.logger.info("%s: %s", self.name, self.feedback_message) + return Status.FAILURE + return None + + def _resolve_current_state( + self, dl: object, participant_id: str + ) -> "tuple[RM, CS_vf | None, CS_d | None] | Status": + """Return current (rm, vf, d) or Status.FAILURE on shape mismatch. + + Wraps resolve_participant_state_from_dl so the try/except lives outside + update(), keeping update()'s McCabe complexity ≤ 10 (C901). + """ + try: + return resolve_participant_state_from_dl( + dl, # type: ignore[arg-type] + participant_id, + ) + except VultronValidationError as exc: + self.feedback_message = ( + f"Participant '{participant_id}' status is not core-shaped:" + f" {exc} (ARCH-15-001)" + ) + self.logger.warning("%s: %s", self.name, self.feedback_message) + return Status.FAILURE def update(self) -> Status: if (f := self._require_datalayer()) is not None: @@ -167,9 +224,10 @@ def update(self) -> Status: # CreateParticipantStatusNode will report this; pass through. return Status.SUCCESS - current_rm, current_vf, current_d = resolve_participant_state_from_dl( - dl, participant_id - ) + state = self._resolve_current_state(dl, participant_id) + if isinstance(state, Status): + return state + current_rm, current_vf, current_d = state participant_obj = dl.read(participant_id) # --- RM dimension --- @@ -216,19 +274,10 @@ def update(self) -> Status: return failure # --- PXA dimension --- - if self._pxa_state is not None and isinstance( - participant_obj, CaseParticipant - ): - current_pxa = _resolve_current_pxa(case, participant_obj) - if self._pxa_state != current_pxa and not is_valid_pxa_transition( - current_pxa, self._pxa_state - ): - self.feedback_message = ( - f"Invalid PXA transition" - f" {current_pxa!r} → {self._pxa_state!r}" - ) - self.logger.info("%s: %s", self.name, self.feedback_message) - return Status.FAILURE + if ( + failure := self._check_pxa_transition(case, participant_obj) + ) is not None: + return failure effective_rm = ( self._rm_state if self._rm_state is not None else current_rm diff --git a/vultron/core/behaviors/case/nodes/vfd_role_guards.py b/vultron/core/behaviors/case/nodes/vfd_role_guards.py index d72ddfafc..0c29dbac8 100644 --- a/vultron/core/behaviors/case/nodes/vfd_role_guards.py +++ b/vultron/core/behaviors/case/nodes/vfd_role_guards.py @@ -26,6 +26,9 @@ MUST NOT hold ``CVDRole.OBSERVER`` as their only role (CM-25-005) - :class:`CheckIsCaseOwnerNode` — hard bypass in ``StatusAdoptionGate``: sender MUST hold ``CVDRole.CASE_OWNER`` (RSH-01-002) + +On-behalf assertion guards (ADR-0084) live in :mod:`on_behalf_guards` and are +re-exported here for backward compatibility. """ import logging diff --git a/vultron/core/predicates/participants.py b/vultron/core/predicates/participants.py index db261d037..c66dec87d 100644 --- a/vultron/core/predicates/participants.py +++ b/vultron/core/predicates/participants.py @@ -21,10 +21,37 @@ delegate to these predicates. """ +from typing import TYPE_CHECKING + from vultron.core.models.case_participant import CaseParticipant from vultron.core.states.rm import RM from vultron.enums.roles import CVDRole +if TYPE_CHECKING: + from vultron.core.states.cs import CS_vf + + +def vendor_vf_invariant_ok( + roles: list[CVDRole], + vf_state: "CS_vf | None", +) -> bool: + """Return True when (roles, vf_state) satisfies the Vendor-implies-V invariant. + + A participant holding ``CVDRole.VENDOR`` is by definition aware of the case; + their VF state MUST NOT be ``CS_vf.vf`` (vendor-unaware). ``None`` means + no VF assertion is being made and is always valid. Non-vendor roles are + unconstrained by this rule. + + Per ADR-0084, PRM-06-002. + """ + if vf_state is None: + return True + if CVDRole.VENDOR not in roles: + return True + from vultron.core.states.cs import CS_vf # avoid circular at module level + + return vf_state != CS_vf.vf + def all_participants_rm_closed( participants: list[CaseParticipant], diff --git a/vultron/core/use_cases/triggers/case/__init__.py b/vultron/core/use_cases/triggers/case/__init__.py index bbe4b49ed..a5bedebd2 100644 --- a/vultron/core/use_cases/triggers/case/__init__.py +++ b/vultron/core/use_cases/triggers/case/__init__.py @@ -23,6 +23,7 @@ from vultron.core.use_cases.triggers.requests import ( AddObjectToCaseTriggerRequest, + AddOnBehalfStatusTriggerRequest, AddParticipantStatusTriggerRequest, AddReportToCaseTriggerRequest, CreateCaseTriggerRequest, @@ -32,6 +33,7 @@ ) from .add_object import SvcAddObjectToCaseUseCase +from .add_on_behalf_status import SvcAddOnBehalfStatusUseCase from .add_participant_status import SvcAddParticipantStatusUseCase from .add_report import SvcAddReportToCaseUseCase from .create import SvcCreateCaseUseCase @@ -41,6 +43,7 @@ __all__ = [ "AddObjectToCaseTriggerRequest", + "AddOnBehalfStatusTriggerRequest", "AddParticipantStatusTriggerRequest", "AddReportToCaseTriggerRequest", "CreateCaseTriggerRequest", @@ -48,6 +51,7 @@ "EngageCaseTriggerRequest", "LeaveCaseTriggerRequest", "SvcAddObjectToCaseUseCase", + "SvcAddOnBehalfStatusUseCase", "SvcAddParticipantStatusUseCase", "SvcAddReportToCaseUseCase", "SvcCreateCaseUseCase", diff --git a/vultron/core/use_cases/triggers/case/add_on_behalf_status.py b/vultron/core/use_cases/triggers/case/add_on_behalf_status.py new file mode 100644 index 000000000..fdfcf40f0 --- /dev/null +++ b/vultron/core/use_cases/triggers/case/add_on_behalf_status.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERTⓇ and CERT Coordination CenterⓇ are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +import logging +from typing import Any, cast + +import py_trees.behaviour + +from vultron.core.behaviors.case.add_on_behalf_status_trigger_tree import ( + add_on_behalf_status_trigger_bt, +) +from vultron.core.states.cs import CS_d, CS_vf +from vultron.core.use_cases.triggers._base import SvcBTTriggerBase +from vultron.core.use_cases.triggers._helpers import ( + resolve_actor, + resolve_case, +) +from vultron.core.use_cases.triggers.requests import ( + AddOnBehalfStatusTriggerRequest, +) +from vultron.enums.roles import CVDRole + +logger = logging.getLogger(__name__) + + +class SvcAddOnBehalfStatusUseCase(SvcBTTriggerBase): + """Assert v→V or d→D on behalf of a notified-but-not-joined vendor/deployer. + + The ``actor_id`` in the request is the *asserting* actor (Case Manager or + Case Owner); ``target_actor_id`` identifies the vendor or deployer whose + awareness or deployment state is being recorded. + + Only ``CS_vf.Vf`` (v→V) and ``CS_d.D`` (d→D) may be asserted on behalf; + ``CS_vf.VF`` (f→F) is rejected at the request layer (ADR-0084, PRM-06-005). + + BT-15-001: the ``ParticipantStatus`` write happens inside the BT via + ``CreateParticipantStatusNode``, not directly in ``execute()``. + """ + + def _prepare(self) -> None: + request = cast(AddOnBehalfStatusTriggerRequest, self._request) + actor = resolve_actor(request.actor_id, self._dl) + self._actor_id = actor.id_ + self._asserting_actor_id = actor.id_ + self._target_actor_id = request.target_actor_id + self._case_id = resolve_case(request.case_id, self._dl).id_ + self._vf_state: CS_vf | None = request.vf_state + self._d_state: CS_d | None = request.d_state + roles: list[CVDRole] = [] + if request.vf_state is not None: + roles.append(CVDRole.VENDOR) + if request.d_state is not None: + roles.append(CVDRole.DEPLOYER) + self._required_roles = roles + + def _build_tree(self) -> py_trees.behaviour.Behaviour: + def _build_activities(case_manager_id: str) -> list[str]: + status_id = self._result_out.get("status_id") + participant_id = self._result_out.get("participant_id") + if not isinstance(status_id, str) or not isinstance( + participant_id, str + ): + raise RuntimeError( + "CreateParticipantStatusNode did not populate result_out" + " before activity_builder was called" + ) + activity_id = self._factory.add_participant_status_to_participant( + status_id=status_id, + participant_id=participant_id, + actor=self._asserting_actor_id, + to=[case_manager_id], + ) + self._result_out["activity_id"] = activity_id + return [activity_id] + + return add_on_behalf_status_trigger_bt( + case_id=self._case_id, + asserting_actor_id=self._asserting_actor_id, + target_actor_id=self._target_actor_id, + required_roles=self._required_roles, + vf_state=self._vf_state, + d_state=self._d_state, + result_out=self._result_out, + activity_builder=_build_activities, + ) + + def _handle_result(self) -> None: + logger.info( + "Actor '%s' asserted on-behalf status for '%s' in case '%s'", + self._asserting_actor_id, + self._target_actor_id, + self._case_id, + ) + + def execute(self) -> dict[str, Any]: + super().execute() + return { + "activity_id": self._result_out.get("activity_id"), + "status_id": self._result_out.get("status_id"), + } diff --git a/vultron/core/use_cases/triggers/requests.py b/vultron/core/use_cases/triggers/requests.py index 81712826d..713e0afaa 100644 --- a/vultron/core/use_cases/triggers/requests.py +++ b/vultron/core/use_cases/triggers/requests.py @@ -14,7 +14,7 @@ from datetime import datetime, timezone -from pydantic import BaseModel, ConfigDict, field_validator +from pydantic import BaseModel, ConfigDict, field_validator, model_validator from vultron.core.models.base import NonEmptyString, UriString from vultron.core.states.cs import CS_d, CS_pxa, CS_vf @@ -251,6 +251,41 @@ class AddParticipantStatusTriggerRequest(CaseTriggerRequest): pxa_state: CS_pxa | None = None +class AddOnBehalfStatusTriggerRequest(CaseTriggerRequest): + """On-behalf v→V / d→D assertion by Case Manager or Case Owner. + + ``actor_id`` is the asserting actor (must hold CASE_MANAGER or CASE_OWNER); + ``target_actor_id`` is the vendor/deployer whose awareness is being recorded. + ``vf_state`` may only be ``CS_vf.Vf`` (v→V); ``CS_vf.VF`` (f→F) is rejected + here because f→F is always self-declared by the Vendor role holder. + + Per ADR-0084, PRM-06-003/004/005. + """ + + target_actor_id: UriString + vf_state: CS_vf | None = None + d_state: CS_d | None = None + + @field_validator("vf_state") + @classmethod + def vf_state_not_fix_ready(cls, v: CS_vf | None) -> CS_vf | None: + if v is not None and v == CS_vf.VF: + raise ValueError( + "f→F (CS_vf.VF) cannot be asserted on behalf of another actor" + " (ADR-0084, PRM-06-005)" + ) + return v + + @model_validator(mode="after") + def at_least_one_dimension(self) -> "AddOnBehalfStatusTriggerRequest": + if self.vf_state is None and self.d_state is None: + raise ValueError( + "at least one of vf_state or d_state must be provided" + " (PRM-06-003/004)" + ) + return self + + class OfferCaseParticipantRoleTriggerRequest(CaseTriggerRequest): """Trigger request to offer a CVDRole to a target Actor in a Case.