From cffa5a870dd89c5bafbad5c3cb34b46ce4025e76 Mon Sep 17 00:00:00 2001 From: Lrd0036 Date: Fri, 9 Oct 2026 23:33:59 -0500 Subject: [PATCH 1/2] docs(trace): correct OCSF walkthrough for tool results Part of #234. Replace the unsupported OCSFLogger snippet and invented event class with an OCSF 1.9.0 Process Activity event for a subprocess self-exit. Copy schema-defined ACS fields with UUIDs, result correlation, and outputs so the walkthrough does not teach an incompatible payload. Follow the OCSF 1.9.0 version decision in #121. Add focused guards that read the published example, check reflexive process semantics and millisecond timestamps, and validate copied ACS fields against the existing schemas without changing the wire contract. Validation on this source snapshot: 911 tests passed, 1 existing filesystem skip; strict MkDocs build passed. The hosted OCSF validator reported 0 errors and 48 missing-recommended-attribute warnings. Desktop and mobile rendering and both section links were checked. The integration-guide repairs and producer/query examples remain follow-up work under #234 and #235. Signed-off-by: Lrd0036 --- docs/acs.md | 110 ++++++++++++++---------- tests/test_ocsf_examples.py | 162 ++++++++++++++++++++++++++++++++++++ 2 files changed, 228 insertions(+), 44 deletions(-) create mode 100644 tests/test_ocsf_examples.py diff --git a/docs/acs.md b/docs/acs.md index 065c720a..97252c8f 100644 --- a/docs/acs.md +++ b/docs/acs.md @@ -448,64 +448,86 @@ See [Conformance Profiles](./spec/conformance.md) for what each profile requires === "OCSF" - ```python - from acs.trace import OCSFLogger - - # Configure OCSF security event logging - security_logger = OCSFLogger( - endpoint="https://siem.example.com/ocsf", - api_key="your-api-key" - ) - - # Attach to agent for security events - acs_agent.attach_logger(security_logger) - - # Execute agent - security events logged automatically - result = await acs_agent.run("Access customer database") - ``` - + This OCSF 1.9.0 example follows the [Trace class mapping](./spec/trace/events.md#ocsf-event-classes). A `run_command` tool launches `/usr/bin/printf hello`, and the host runtime observes its subprocess exit with code `0` at `2026-10-09T12:00:00Z`. + + The `steps/toolCallResult` trace uses Process Activity (`1007`). This is a reflexive [Terminate activity](https://schema.ocsf.io/classes/process_activity): `printf` exits on its own, so both `actor.process` and `process` identify `printf`. Each records the Python runtime as `parent_process`. The identifiers below are sample values for this scenario. + ```json { - "activity_id": 1, - "activity_name": "Agent Execution", - "category_uid": 3, - "category_name": "Application Activity", - "class_uid": 3001, - "class_name": "AI Agent Activity", + "activity_id": 2, + "activity_name": "Terminate", + "category_uid": 1, + "category_name": "System Activity", + "class_uid": 1007, + "class_name": "Process Activity", + "type_uid": 100702, "severity_id": 1, - "time": 1705318200, + "time": 1791547200000, + "status_id": 1, + "status": "Success", "metadata": { - "version": "1.0.0", + "version": "1.9.0", "product": { - "name": "ACS Agent", + "name": "ACS trace example", "vendor_name": "Example Corp" } }, "actor": { - "session": { - "uid": "sess_123", - "created_time": 1705318200 - }, - "idp": { - "name": "research-assistant", - "uid": "agent_456" + "process": { + "pid": 4243, + "name": "printf", + "cmd_line": "/usr/bin/printf hello", + "terminated_time": 1791547200000, + "parent_process": { + "pid": 4242, + "name": "python", + "cmd_line": "python agent.py" + } } }, - "acs_extensions": { - "reasoning_chain": [ - { - "step": 1, - "thought": "User wants database access", - "action": "evaluate_permissions" - } - ], - "tools_invoked": ["database_query"], - "data_accessed": ["customers.personal_info"], - "risk_score": 8.5 + "device": { + "uid": "host_001", + "hostname": "agent-host.example", + "type_id": 1, + "type": "Server" + }, + "process": { + "pid": 4243, + "name": "printf", + "cmd_line": "/usr/bin/printf hello", + "terminated_time": 1791547200000, + "parent_process": { + "pid": 4242, + "name": "python", + "cmd_line": "python agent.py" + } + }, + "exit_code": 0, + "unmapped": { + "acs": { + "method": "steps/toolCallResult", + "request_id": "1287ce09-fcda-4588-a5e9-3fa83bfe5291", + "agent_id": "agent_456", + "session_id": "7673ad6b-b5b3-4eec-8f46-66c83eef0b52", + "tool": { + "name": "run_command" + }, + "exit_status": "success", + "request_id_ref": "e1572c78-689e-4bdb-91d4-6e019b416849", + "outputs": [ + { "value": "hello" } + ] + } } } ``` + The host runtime supplies `actor.process`, `device`, `process`, the OS `exit_code`, and the occurrence time in milliseconds. These facts supplement the ACS payload; a generic tool call may not have subprocess details. + + `unmapped.acs` copies the request's `method` and `request_id`, the envelope metadata's `agent_id` and `session_id`, and fields from the [tool-call result payload](./spec/instrument/hooks.md#toolcallresult). `request_id_ref` links the result to its earlier `toolCallRequest`. The request and session identifiers are UUIDs. + + This example shows one step event with no attached provenance. Its Guardian decision is recorded as a separate trace event. + === "Real-time Streaming" ```typescript @@ -785,4 +807,4 @@ See [Conformance Profiles](./spec/conformance.md) for what each profile requires - [Core concepts](./topics/core_concepts.md) - [Instrument](./spec/instrument/README.md) - [Trace](./spec/trace/README.md) -- [Inspect](./spec/inspect/README.md) \ No newline at end of file +- [Inspect](./spec/inspect/README.md) diff --git a/tests/test_ocsf_examples.py b/tests/test_ocsf_examples.py new file mode 100644 index 00000000..05c09deb --- /dev/null +++ b/tests/test_ocsf_examples.py @@ -0,0 +1,162 @@ +"""Keep the walkthrough's process example aligned with its stated scenario. + +The OCSF tab previously used an invented class, a seconds timestamp, and the +agent's identity as an identity provider. MkDocs and the link guards accepted +it. Read the published JSON itself so those mistakes cannot return unnoticed. + +These are focused documentation checks, not a complete OCSF validator. The +required process context follows OCSF 1.9.0's Process Activity, System Activity, +Process, and Device definitions at schema commit +856d462bd20dc46cc1ffed2dfffe3b91ef0fbeba. Upstream event validation is a separate +check before publication. +""" + +import json +import re +from copy import deepcopy +from datetime import datetime +from pathlib import Path + +import pytest +from jsonschema import Draft202012Validator, FormatChecker, ValidationError +from referencing import Registry, Resource +from referencing.jsonschema import DRAFT202012 + +ROOT = Path(__file__).resolve().parent.parent +SPEC = ROOT / "specification" / "v0.1.0" + + +def walkthrough_tab(): + source = (ROOT / "docs" / "acs.md").read_text(encoding="utf-8") + return source.partition(' === "OCSF"\n')[2].partition("\n === ")[0] + + +def walkthrough_event(): + blocks = re.findall(r"```json\n(.*?)```", walkthrough_tab(), re.DOTALL) + assert len(blocks) == 1, "The OCSF tab must contain one complete event example" + return json.loads(blocks[0]) + + +def declared_objects_only(schema): + """Reject invented wrapper fields in this example, leaving output values opaque. + + The wire schemas remain extensible. This constraint applies only to the + documentation's claim to copy existing ACS fields. + """ + if isinstance(schema, list): + return [declared_objects_only(item) for item in schema] + if not isinstance(schema, dict): + return schema + result = {key: declared_objects_only(value) for key, value in schema.items()} + if schema.get("type") == "object" and "properties" in schema: + result["additionalProperties"] = False + return result + + +def validate_copied_acs_fields(acs): + envelope = json.loads((SPEC / "request-envelope.json").read_text(encoding="utf-8")) + hook = json.loads((SPEC / "hooks" / "tool-call-result.json").read_text(encoding="utf-8")) + metadata = envelope["$defs"]["Metadata"] + payload_fields = hook["properties"] + metadata_fields = metadata["properties"] + allowed = set(payload_fields) | set(metadata_fields) | {"method", "request_id"} + assert not set(acs) - allowed, f"Undeclared ACS fields: {set(acs) - allowed}" + + resources = [] + for path in sorted(SPEC.rglob("*.json")): + schema = json.loads(path.read_text(encoding="utf-8")) + if isinstance(schema, dict) and "$id" in schema: + resources.append((schema["$id"], Resource.from_contents(schema, default_specification=DRAFT202012))) + registry = Registry().with_resources(resources) + + for schema, value in ( + (metadata, {key: value for key, value in acs.items() if key in metadata_fields}), + (hook, {key: value for key, value in acs.items() if key in payload_fields}), + (envelope["properties"]["method"], acs["method"]), + (envelope["$defs"]["AcsParams"]["properties"]["request_id"], acs["request_id"]), + ): + Draft202012Validator( + declared_objects_only(schema), registry=registry, format_checker=FormatChecker() + ).validate(value) + + assert acs["method"] == "steps/toolCallResult" + assert acs["request_id"] != acs["request_id_ref"] + + +def test_walkthrough_uses_the_mapped_process_class(): + event = walkthrough_event() + method = event["unmapped"]["acs"].get("method") + assert method == "steps/toolCallResult" + mapping = json.loads( + (SPEC / "trace" / "ocsf-mapping.json") + .read_text(encoding="utf-8") + )["properties"]["step_to_class"]["default"][method] + + assert event["class_uid"] == mapping["class_uid"] + assert event["class_name"] == mapping["class_name"] + assert event["category_uid"] == 1 # System Activity + assert event["category_name"] == "System Activity" + assert event["activity_id"] == 2 # A subprocess exiting on its own + assert event["activity_name"] == "Terminate" + assert event["type_uid"] == event["class_uid"] * 100 + event["activity_id"] + assert event["metadata"]["version"] == "1.9.0" + assert event["metadata"]["product"]["name"] + assert "ocsf" not in event["metadata"] + + +def test_walkthrough_represents_a_reflexive_process_exit(): + event = walkthrough_event() + assert "process" in event["actor"], "The actor must identify the exiting process" + assert "device" in event, "Process Activity requires the actual host device" + assert "process" in event, "The tool's subprocess must be identified" + assert event["actor"]["process"] == event["process"], ( + "A self-exit's actor and target must both identify the exiting process" + ) + assert type(event["process"]["pid"]) is int + assert event["process"]["pid"] > 0 + parent = event["process"]["parent_process"] + assert type(parent["pid"]) is int and parent["pid"] > 0 + assert parent["pid"] != event["process"]["pid"] + assert type(event["device"]["type_id"]) is int + assert "idp" not in event["actor"] + + acs = event["unmapped"]["acs"] + assert event["device"]["uid"] != acs["agent_id"] + assert type(event["exit_code"]) is int + assert event["exit_code"] == 0 and acs["exit_status"] == "success" + assert "acs_extensions" not in event + assert "asop" not in event["unmapped"] + + +def test_walkthrough_copies_declared_acs_fields(): + validate_copied_acs_fields(walkthrough_event()["unmapped"]["acs"]) + + +@pytest.mark.parametrize("path,value", [ + (("tool", "id"), "run_command"), + (("tool", "execution_id"), "exec_123"), + (("step",), {"id": "step_result_1", "type": "toolCallResult"}), + (("session_id",), "sess_123"), + (("request_id",), "step_result_1"), + (("request_id_ref",), "exec_123"), +]) +def test_copied_acs_fields_reject_old_aliases_and_non_uuid_ids(path, value): + acs = deepcopy(walkthrough_event()["unmapped"]["acs"]) + validate_copied_acs_fields(acs) + target = acs + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + with pytest.raises((AssertionError, ValidationError)): + validate_copied_acs_fields(acs) + + +def test_walkthrough_timestamps_match_the_documented_exit(): + event = walkthrough_event() + timestamp = re.search(r"at `([^`]+Z)`", walkthrough_tab()) + assert timestamp, "The scenario must state its occurrence time in UTC" + exited_at = datetime.fromisoformat(timestamp[1]) + expected_ms = int(exited_at.timestamp() * 1000) + assert event["time"] == expected_ms, "OCSF occurrence time uses milliseconds" + assert event["process"]["terminated_time"] == expected_ms + assert event["actor"]["process"]["terminated_time"] == expected_ms From 147f7d4f6453011761b657af21a8abaf5dc9353e Mon Sep 17 00:00:00 2001 From: Lrd0036 Date: Wed, 4 Nov 2026 12:14:24 -0600 Subject: [PATCH 2/2] Set the OCSF walkthrough's correlation_uid and check it Since #275 the mapping requires every step event to set metadata.correlation_uid to its own :, so a SIEM can join the step to its decision event on native fields. The example now carries it, and the guard fails if it is dropped, reversed, built from request_id_ref, or uses another separator. Signed-off-by: Lrd0036 --- docs/acs.md | 5 +++-- tests/test_ocsf_examples.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/docs/acs.md b/docs/acs.md index 7b70bb3b..6b1bb8c6 100644 --- a/docs/acs.md +++ b/docs/acs.md @@ -472,7 +472,8 @@ See [Conformance Profiles](./spec/conformance.md) for what each profile requires "product": { "name": "ACS trace example", "vendor_name": "Example Corp" - } + }, + "correlation_uid": "7673ad6b-b5b3-4eec-8f46-66c83eef0b52:1287ce09-fcda-4588-a5e9-3fa83bfe5291" }, "actor": { "process": { @@ -528,7 +529,7 @@ See [Conformance Profiles](./spec/conformance.md) for what each profile requires `unmapped.acs` copies the request's `method` and `request_id`, the envelope metadata's `agent_id` and `session_id`, and fields from the [tool-call result payload](./spec/instrument/hooks.md#toolcallresult). `request_id_ref` links the result to its earlier `toolCallRequest`. The request and session identifiers are UUIDs. - This example shows one step event with no attached provenance. Its Guardian decision is recorded as a separate trace event. + `metadata.correlation_uid` repeats the step's pair as `:`, because SIEM content reads native fields, not `unmapped`. The Guardian's decision is a separate trace event with the same `correlation_uid`, so a SIEM can join the two on that one field. This example shows the step event only, with no attached provenance. === "Real-time Streaming" diff --git a/tests/test_ocsf_examples.py b/tests/test_ocsf_examples.py index 05c09deb..ac485e61 100644 --- a/tests/test_ocsf_examples.py +++ b/tests/test_ocsf_examples.py @@ -132,6 +132,35 @@ def test_walkthrough_copies_declared_acs_fields(): validate_copied_acs_fields(walkthrough_event()["unmapped"]["acs"]) +def check_correlation_uid(event): + """Every OCSF step event sets metadata.correlation_uid to its own pair (step_to_class in the mapping).""" + acs = event["unmapped"]["acs"] + assert event["metadata"].get("correlation_uid") == f"{acs['session_id']}:{acs['request_id']}", ( + "metadata.correlation_uid must be the step's :" + ) + + +def test_walkthrough_sets_the_step_correlation_uid(): + check_correlation_uid(walkthrough_event()) + + +@pytest.mark.parametrize("make", [ + None, + lambda acs: f"{acs['request_id']}:{acs['session_id']}", + lambda acs: f"{acs['session_id']}:{acs['request_id_ref']}", + lambda acs: f"{acs['session_id']}/{acs['request_id']}", +], ids=["dropped", "reversed", "gated_request", "separator"]) +def test_correlation_uid_rejects_a_dropped_or_wrong_pair(make): + event = walkthrough_event() + check_correlation_uid(event) + if make is None: + del event["metadata"]["correlation_uid"] + else: + event["metadata"]["correlation_uid"] = make(event["unmapped"]["acs"]) + with pytest.raises(AssertionError): + check_correlation_uid(event) + + @pytest.mark.parametrize("path,value", [ (("tool", "id"), "run_command"), (("tool", "execution_id"), "exec_123"),