diff --git a/README.md b/README.md index 020bcf2..2f2decf 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,7 @@ as structured, serializable data. An investigation contains observables and their relationships, Findings, supporting Evidence, threat intelligence, enrichments, and tags. Cyvest maintains -deterministic keys, calculates scores and security levels, and records mutations -in an audit log. +deterministic keys and calculates scores and security levels. ## Main capabilities @@ -79,8 +78,7 @@ cv.io_save_json("investigation.json") ``` Public model objects are exposed through read-only proxies. Mutations use the -facade or fluent methods so score propagation, reverse links, and the audit log -remain consistent. +facade or fluent methods so score propagation and reverse links remain consistent. For deterministic reports and comparisons, pass an explicit `investigation_id` when creating the investigation. @@ -361,16 +359,6 @@ cv = Cyvest(score_mode_obs=ScoreMode.SUM) # accumulative children - `Investigation.investigation_id` is a stable ULID included in exports. - Findings keep a *canonical origin* (`origin_investigation_id`) for LOCAL_ONLY propagation; it is compared against the current investigation id. -**Audit log** - -- All meaningful changes (including score/level changes) are recorded in the investigation-level audit log. -- Per-object histories are not stored; use `cv.investigation_get_audit_log()` to review changes. -- For compact, deterministic JSON output (useful for testing/diffing), exclude the audit log: - ```python - cv.io_save_json("output.json", include_audit_log=False) # audit_log: null - cv.io_to_invest(include_audit_log=False) # schema.audit_log is None - ``` - To force cross-investigation propagation for a specific link, use a GLOBAL link: ```python @@ -712,9 +700,8 @@ The repo includes a PNPM workspace under `js/` with three packages: - `@cyvest/cyvest-app`: Vite demo that bundles the JS packages with sample investigations. The JS packages track the generated schema; serialized investigations should include fields like -`investigation_id`, `investigation_name`, `audit_log`, `score_display`, `finding_links`, and -`observable_links`. The investigation start time is recorded as an `INVESTIGATION_STARTED` event -in the `audit_log`. +`investigation_id`, `investigation_name`, `score_display`, `finding_links`, and +`observable_links`. See `docs/js-packages.md` for workspace commands and usage snippets. diff --git a/docs/getting-started/concepts.md b/docs/getting-started/concepts.md index 6aa2d52..1314649 100644 --- a/docs/getting-started/concepts.md +++ b/docs/getting-started/concepts.md @@ -46,7 +46,7 @@ Each observable has: > instances rather than raw dataclasses. These proxies provide live scores/levels but raise an error if you > attempt to assign attributes. All mutations flow through the Investigation layer, so use the facade helpers > (`cv.observable_add_threat_intel`, `cv.observable_set_level`, …) or the fluent methods on the proxies themselves -> (`with_ti`, `relate_to`, `link_observable`, `set_level`, etc.) so the score engine and audit log remain consistent. +> (`with_ti`, `relate_to`, `link_observable`, `set_level`, etc.) so the score engine remains consistent. > Safe metadata fields (`comment`, `extra`, `internal`, etc.) can be updated via the dedicated `update_metadata()` > helpers on each proxy. Use `set_level()` to update the level without changing the score: > @@ -371,36 +371,6 @@ explicit direction leaves both scores untouched. This is intended: a correlation with no established deduction mechanism has no reason to make a score propagate. `EXTRACTION` and `PIVOT` default to `OUTBOUND` and do propagate. -### Audit Log - -All meaningful changes are recorded in a centralized, append-only audit log at the investigation level: - -```python -# Observable score changes -obs = cv.observable_create(cv.OBS.IPV4, "10.0.0.1") -cv.observable_add_threat_intel(obs.key, "source1", score=Decimal("5.0")) -cv.observable_add_threat_intel(obs.key, "source2", score=Decimal("8.0")) - -events = cv.investigation_get_audit_log() -obs_score_events = [ - event - for event in events - if event.object_key == obs.key and event.event_type.startswith("SCORE") -] -for event in obs_score_events: - print(event.timestamp, event.details["old_score"], "→", event.details["new_score"]) - print("Level:", event.details["old_level"], "→", event.details["new_level"]) - print("Reason:", event.reason) -``` - -**Audit Event Fields (score changes):** - -- Timestamp -- Old/new score values -- Old/new level values -- Reason for change (threat intel, propagation, merge, manual, etc.) -- Contributing investigation IDs when relevant (e.g., merges) - Investigation names are optional, human-readable labels. They are serialized separately from `investigation_id` and are never used for scoring or propagation. ### Hierarchical Score Propagation diff --git a/docs/getting-started/quickstart.md b/docs/getting-started/quickstart.md index 86dd19a..c129f63 100644 --- a/docs/getting-started/quickstart.md +++ b/docs/getting-started/quickstart.md @@ -166,9 +166,6 @@ cv.io_save_json("investigation.json") cv.io_save_markdown("report.md") # Hide observables while keeping aggregate stats/whitelists cv.io_save_markdown("redacted_report.md", include_observables=False) - -# For compact, deterministic JSON (useful for testing/diffing): -cv.io_save_json("deterministic.json", include_audit_log=False) ``` !!! tip "Filtering findings by severity" @@ -181,7 +178,7 @@ cv.io_save_json("deterministic.json", include_audit_log=False) The docs assume you write to the project root, but automation pipelines typically point to `dist/` (JSON) and `reports/` (Markdown/PDF). Adjust paths to match your workflow. !!! note "Provenance fields in JSON" - Exports include `investigation_id`, optional `investigation_name`, and the investigation-level `audit_log`, plus finding origins (`origin_investigation_id`) and link fields (`observable_links`, `finding_links`) needed for scoring after merges. + Exports include `investigation_id`, optional `investigation_name`, plus finding origins (`origin_investigation_id`) and link fields (`observable_links`, `finding_links`) needed for scoring after merges. --- diff --git a/docs/index.md b/docs/index.md index b398dca..1628d2a 100644 --- a/docs/index.md +++ b/docs/index.md @@ -25,7 +25,7 @@ Build, score, and narrate cybersecurity investigations with a single fluent Pyth | Area | Why it matters | What to look at | | --- | --- | --- | | **Structured objects** | Model observables, findings, TI, tags, and enrichments with typed helpers | `cyvest.model`, [Concepts](getting-started/concepts.md#observables) | -| **Deterministic scoring** | MAX/SUM propagation, centralized audit log, and automatic level classification | `cyvest.score`, [Scoring System](getting-started/concepts.md#scoring-system) | +| **Deterministic scoring** | MAX/SUM propagation and automatic level classification | `cyvest.score`, [Scoring System](getting-started/concepts.md#scoring-system) | | **Fluent helpers** | Builder-style methods with deterministic keys and safe merges | `cyvest.cyvest`, [Quick Start](getting-started/quickstart.md#using-the-fluent-api) | | **Shared context** | Thread-safe fragments that can reconcile into a single story | `cyvest.shared.SharedInvestigationContext`, [Guide](shared-investigation-context.md) | | **Comparison** | Compare investigations with tolerance rules for regression testing | `cyvest.compare`, [Guide](comparing-investigations.md) | diff --git a/docs/js-packages.md b/docs/js-packages.md index 6b95f98..d59b362 100644 --- a/docs/js-packages.md +++ b/docs/js-packages.md @@ -3,9 +3,8 @@ Cyvest ships a small JavaScript/TypeScript workspace alongside the Python API. Use these packages to validate serialized investigations, power UI integrations, or explore the data model in a browser. The JS packages follow the generated schema. Serialized investigations should include the -schema-required fields such as `investigation_id`, `investigation_name`, `audit_log`, -`score_display`, `finding_links`, and `observable_links`. The investigation start time is -recorded as an `INVESTIGATION_STARTED` event in the `audit_log`. +schema-required fields such as `investigation_id`, `investigation_name`, +`score_display`, `finding_links`, and `observable_links`. ## Packages diff --git a/docs/migration-v5-to-v6.md b/docs/migration-v5-to-v6.md index fd5aeb7..30b3c06 100644 --- a/docs/migration-v5-to-v6.md +++ b/docs/migration-v5-to-v6.md @@ -25,7 +25,7 @@ This guide covers both serialized investigations and Python integrations. | No required schema version | Exact `schema_version: "6.0.0"` | The scoring model, relationship propagation, tags, enrichments, threat -intelligence, audit log, and immutable proxy behavior remain in place. +intelligence, and immutable proxy behavior remain in place. ## 1. Upgrade Cyvest @@ -63,7 +63,7 @@ The migration command: - renames the root `checks` collection to `findings`; - renames `check_name` to `finding_name`; - rewrites `chk:` keys to `fnd:` keys; -- updates Finding references in tags and audit events; +- updates Finding references in tags; - recalculates observable and threat-intelligence keys; - initializes `evidences` and every `evidence_links` collection; - preserves existing `EMAIL` observables; @@ -97,7 +97,7 @@ assert cv.finding_get_all() ``` Do not update keys with a global text replacement. Observable, threat -intelligence, tag, and audit references must remain consistent; the migration +intelligence, and tag references must remain consistent; the migration command performs those updates as one validated operation. ## 3. Rename Check APIs to Finding APIs diff --git a/examples/04_email.py b/examples/04_email.py index 86b1b9d..061c7d6 100644 --- a/examples/04_email.py +++ b/examples/04_email.py @@ -489,16 +489,8 @@ def run(self, cy: Cyvest) -> None: @click_logger_params @click.option("-w", "--workers", type=int, default=1) @click.option("--stats", "stats", is_flag=True, default=False) -@click.option("--audit", "audit", is_flag=True, default=False) -@click.option( - "--no-audit-log", - "no_audit_log", - is_flag=True, - default=False, - help="Exclude audit log from JSON output for deterministic output", -) @click.option("-o", "--output", type=click.Path(dir_okay=False, path_type=Path), default=None) -def main(workers, stats, audit, no_audit_log, output): +def main(workers, stats, output): """Main execution demonstrating multi-threaded investigation.""" # Prepare input data @@ -553,13 +545,13 @@ def main(workers, stats, audit, no_audit_log, output): # Display results logger.info("Investigation complete - displaying summary - score should be 36.1") - cy.display_summary(show_audit_log=audit) + cy.display_summary() if stats: cy.display_statistics() if output is not None: logger.info("[bold cyan]Generating json...[/bold cyan]") - json_path = cy.io_save_json(output, include_audit_log=not no_audit_log) + json_path = cy.io_save_json(output) size_kb = Path(json_path).stat().st_size / 1024 logger.info("[green]✓ Full json saved to: %s (%.2f KB)[/green]", json_path, size_kb) diff --git a/examples/05_graph_dataset.py b/examples/05_graph_dataset.py index 50391dc..37f7a8f 100644 --- a/examples/05_graph_dataset.py +++ b/examples/05_graph_dataset.py @@ -20,15 +20,8 @@ @click.command() @click_logger_params -@click.option( - "--no-audit-log", - "no_audit_log", - is_flag=True, - default=False, - help="Exclude audit log from JSON output for deterministic output", -) @click.option("-o", "--output", type=click.Path(dir_okay=False, path_type=Path), default=None) -def main(no_audit_log: bool = False, output: Path | None = None) -> None: +def main(output: Path | None = None) -> None: # Create a comprehensive investigation with various observables cv = Cyvest( investigation_id="cyvest-visual-example", @@ -159,7 +152,7 @@ def main(no_audit_log: bool = False, output: Path | None = None) -> None: # Generate json logger.info("[bold cyan]Generating json...[/bold cyan]") output_path = output or (Path(tempfile.gettempdir()) / "cyvest_investigation.json") - json_path = cv.io_save_json(output_path, include_audit_log=not no_audit_log) + json_path = cv.io_save_json(output_path) size_kb = Path(json_path).stat().st_size / 1024 logger.info("[green]✓ Full json saved to: %s (%.2f KB)[/green]", json_path, size_kb) diff --git a/js/packages/cyvest-app/src/App.tsx b/js/packages/cyvest-app/src/App.tsx index b930593..ac67acf 100644 --- a/js/packages/cyvest-app/src/App.tsx +++ b/js/packages/cyvest-app/src/App.tsx @@ -1,5 +1,4 @@ import type { CyvestInvestigation } from "@cyvest/cyvest-js"; -import { getStartedAt } from "@cyvest/cyvest-js"; import { CyvestGraph, DARK_CYVEST_THEME, @@ -75,8 +74,7 @@ export const App: React.FC = () => { investigation.investigation_id}

- Started {getStartedAt(investigation) ?? "N/A"} · Schema{" "} - {investigation.schema_version} + Schema {investigation.schema_version}

diff --git a/js/packages/cyvest-app/src/investigations/cyvest_email.json b/js/packages/cyvest-app/src/investigations/cyvest_email.json index 1eaab1d..bb1034c 100644 --- a/js/packages/cyvest-app/src/investigations/cyvest_email.json +++ b/js/packages/cyvest-app/src/investigations/cyvest_email.json @@ -6,7 +6,6 @@ "level": "MALICIOUS", "whitelisted": false, "whitelists": [], - "audit_log": null, "observables": { "obs:artifact:root": { "type": "artifact", diff --git a/js/packages/cyvest-app/src/investigations/cyvest_visual.json b/js/packages/cyvest-app/src/investigations/cyvest_visual.json index 376a7ee..80d39d1 100644 --- a/js/packages/cyvest-app/src/investigations/cyvest_visual.json +++ b/js/packages/cyvest-app/src/investigations/cyvest_visual.json @@ -6,7 +6,6 @@ "level": "MALICIOUS", "whitelisted": false, "whitelists": [], - "audit_log": null, "observables": { "obs:file:root": { "type": "file", diff --git a/js/packages/cyvest-js/src/getters.ts b/js/packages/cyvest-js/src/getters.ts index 503d2a7..4a03e00 100644 --- a/js/packages/cyvest-js/src/getters.ts +++ b/js/packages/cyvest-js/src/getters.ts @@ -379,29 +379,6 @@ export function getCounts(inv: CyvestInvestigation): InvestigationCounts { }; } -/** - * Get the investigation start time from the event log. - * - * Looks for the INVESTIGATION_STARTED event and returns its timestamp. - * - * @param inv - The investigation - * @returns The start timestamp string or undefined if not found - * - * @example - * ```ts - * const startedAt = getStartedAt(investigation); - * if (startedAt) { - * console.log(`Started: ${startedAt}`); - * } - * ``` - */ -export function getStartedAt(inv: CyvestInvestigation): string | undefined { - const event = inv.audit_log?.find( - (e) => e.event_type === "INVESTIGATION_STARTED" - ); - return event?.timestamp; -} - // ============================================================================ // Tag Aggregation // ============================================================================ diff --git a/js/packages/cyvest-js/src/types.generated.ts b/js/packages/cyvest-js/src/types.generated.ts index 3f75756..93fa44b 100644 --- a/js/packages/cyvest-js/src/types.generated.ts +++ b/js/packages/cyvest-js/src/types.generated.ts @@ -15,15 +15,6 @@ export type Justification = string | null; * List of whitelist entries applied to this investigation. */ export type Whitelists = InvestigationWhitelist[]; -/** - * Append-only investigation audit log. Null when serialization disabled audit. - */ -export type AuditLog = AuditEvent[] | null; -export type Actor = string | null; -export type Reason = string | null; -export type Tool = string | null; -export type ObjectType = string | null; -export type ObjectKey = string | null; export type Subtype = string | null; export type Namespace = string | null; export type Subtype1 = string | null; @@ -89,7 +80,6 @@ export interface CyvestInvestigation { */ whitelisted: boolean; whitelists: Whitelists; - audit_log?: AuditLog; observables: Observables; findings: Findings; evidences: Evidences; @@ -112,24 +102,6 @@ export interface InvestigationWhitelist { justification?: Justification; [k: string]: unknown; } -/** - * Centralized audit event for investigation-level changes. - */ -export interface AuditEvent { - event_id: string; - timestamp: string; - event_type: string; - actor?: Actor; - reason?: Reason; - tool?: Tool; - object_type?: ObjectType; - object_key?: ObjectKey; - details?: Details; - [k: string]: unknown; -} -export interface Details { - [k: string]: unknown; -} /** * Observables keyed by their unique key. */ diff --git a/js/packages/cyvest-js/tests/getters-finders.test.ts b/js/packages/cyvest-js/tests/getters-finders.test.ts index 5a7b8ae..6a62026 100644 --- a/js/packages/cyvest-js/tests/getters-finders.test.ts +++ b/js/packages/cyvest-js/tests/getters-finders.test.ts @@ -18,7 +18,6 @@ import { getAllTags, getAllObservables, getCounts, - getStartedAt, getTagChildren, getTagDescendants, getTagAggregatedScore, @@ -51,15 +50,6 @@ function createTestInvestigation(): CyvestInvestigation { score_display: "7.50", level: "MALICIOUS", whitelisted: false, - audit_log: [ - { - event_id: "01HXYZTESTEVENT001", - timestamp: "2024-01-01T00:00:00Z", - event_type: "INVESTIGATION_STARTED", - object_type: "investigation", - object_key: "01HXYZTESTINVESTIGATION", - }, - ], whitelists: [ { identifier: "wl-1", @@ -362,25 +352,6 @@ describe("Getters", () => { expect(counts.whitelists).toBe(1); }); }); - - describe("getStartedAt", () => { - it("returns timestamp from INVESTIGATION_STARTED event", () => { - const startedAt = getStartedAt(inv); - expect(startedAt).toBe("2024-01-01T00:00:00Z"); - }); - - it("returns undefined when no audit_log", () => { - const invWithoutAuditLog = { ...inv, audit_log: undefined }; - const startedAt = getStartedAt(invWithoutAuditLog); - expect(startedAt).toBeUndefined(); - }); - - it("returns undefined when no INVESTIGATION_STARTED event", () => { - const invWithEmptyLog = { ...inv, audit_log: [] }; - const startedAt = getStartedAt(invWithEmptyLog); - expect(startedAt).toBeUndefined(); - }); - }); }); describe("Finders", () => { diff --git a/js/packages/cyvest-js/tests/graph.test.ts b/js/packages/cyvest-js/tests/graph.test.ts index ccb5768..8dc152c 100644 --- a/js/packages/cyvest-js/tests/graph.test.ts +++ b/js/packages/cyvest-js/tests/graph.test.ts @@ -26,15 +26,6 @@ function createGraphTestInvestigation(): CyvestInvestigation { score_display: "5.00", level: "MALICIOUS", whitelisted: false, - audit_log: [ - { - event_id: "01HXYZTESTEVENT001", - timestamp: "2024-01-01T00:00:00Z", - event_type: "INVESTIGATION_STARTED", - object_type: "investigation", - object_key: "01HXYZGRAPHINVESTIGATION", - }, - ], whitelists: [], observables: { "obs:file:msg1": { diff --git a/schema/cyvest.schema.json b/schema/cyvest.schema.json index d67d6d7..2e56204 100644 --- a/schema/cyvest.schema.json +++ b/schema/cyvest.schema.json @@ -1,96 +1,5 @@ { "$defs": { - "AuditEvent": { - "additionalProperties": true, - "description": "Centralized audit event for investigation-level changes.", - "properties": { - "event_id": { - "title": "Event Id", - "type": "string" - }, - "timestamp": { - "format": "date-time", - "title": "Timestamp", - "type": "string" - }, - "event_type": { - "title": "Event Type", - "type": "string" - }, - "actor": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "title": "Actor" - }, - "reason": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "title": "Reason" - }, - "tool": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "title": "Tool" - }, - "object_type": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "title": "Object Type" - }, - "object_key": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "title": "Object Key" - }, - "details": { - "additionalProperties": true, - "title": "Details", - "type": "object" - } - }, - "required": [ - "event_id", - "timestamp", - "event_type" - ], - "title": "AuditEvent", - "type": "object" - }, "DataExtractionSchema": { "additionalProperties": false, "description": "Schema for data extraction metadata.", @@ -927,21 +836,6 @@ "title": "Whitelists", "type": "array" }, - "audit_log": { - "anyOf": [ - { - "items": { - "$ref": "#/$defs/AuditEvent" - }, - "type": "array" - }, - { - "type": "null" - } - ], - "description": "Append-only investigation audit log. Null when serialization disabled audit.", - "title": "Audit Log" - }, "observables": { "additionalProperties": { "$ref": "#/$defs/Observable" diff --git a/scripts/generate.sh b/scripts/generate.sh index 47304f9..592840e 100755 --- a/scripts/generate.sh +++ b/scripts/generate.sh @@ -3,7 +3,7 @@ uv run cyvest schema -o ./schema/cyvest.schema.json pnpm -C js/packages/cyvest-js run generate:types -uv run examples/04_email.py --no-audit-log -o ./js/packages/cyvest-app/src/investigations/cyvest_email.json -uv run examples/05_graph_dataset.py --no-audit-log -o ./js/packages/cyvest-app/src/investigations/cyvest_visual.json +uv run examples/04_email.py -o ./js/packages/cyvest-app/src/investigations/cyvest_email.json +uv run examples/05_graph_dataset.py -o ./js/packages/cyvest-app/src/investigations/cyvest_visual.json pnpm -C js run build pnpm -C js run test:ci diff --git a/src/cyvest/cyvest.py b/src/cyvest/cyvest.py index 794377c..8d06c76 100644 --- a/src/cyvest/cyvest.py +++ b/src/cyvest/cyvest.py @@ -200,10 +200,6 @@ def investigation_set_name(self, name: str | None, reason: str | None = None) -> """Set or clear the human-readable investigation name.""" self._investigation.set_investigation_name(name, reason=reason) - def investigation_get_audit_log(self) -> tuple: - """Return the investigation-level audit log.""" - return tuple(self._investigation.get_audit_log()) - def investigation_add_whitelist( self, identifier: str, name: str, justification: str | None = None ) -> InvestigationWhitelist: @@ -1204,7 +1200,7 @@ def get_statistics(self) -> StatisticsSchema: # Serialization and I/O methods - def io_save_json(self, filepath: str | Path, *, include_audit_log: bool = True) -> str: + def io_save_json(self, filepath: str | Path) -> str: """ Save the investigation to a JSON file. @@ -1212,8 +1208,6 @@ def io_save_json(self, filepath: str | Path, *, include_audit_log: bool = True) Args: filepath: Path to save the JSON file (relative or absolute) - include_audit_log: Include audit log in output (default: True). - When False, audit_log is set to null for compact, deterministic output. Returns: Absolute path to the saved file as a string @@ -1226,10 +1220,8 @@ def io_save_json(self, filepath: str | Path, *, include_audit_log: bool = True) >>> cv = Cyvest() >>> path = cv.io_save_json("investigation.json") >>> print(path) # /absolute/path/to/investigation.json - >>> # For compact, deterministic output: - >>> path = cv.io_save_json("output.json", include_audit_log=False) """ - save_investigation_json(self._investigation, filepath, include_audit_log=include_audit_log) + save_investigation_json(self._investigation, filepath) return str(Path(filepath).resolve()) def io_save_markdown( @@ -1297,14 +1289,10 @@ def io_to_markdown( self._investigation, include_tags, include_enrichments, include_observables, exclude_levels ) - def io_to_invest(self, *, include_audit_log: bool = True) -> InvestigationSchema: + def io_to_invest(self) -> InvestigationSchema: """ Serialize the investigation to an InvestigationSchema. - Args: - include_audit_log: Include audit log in serialization (default: True). - When False, audit_log is set to None for compact, deterministic output. - Returns: InvestigationSchema instance (use .model_dump() for dict) @@ -1313,20 +1301,13 @@ def io_to_invest(self, *, include_audit_log: bool = True) -> InvestigationSchema >>> schema = cv.io_to_invest() >>> print(schema.score, schema.level) >>> dict_data = schema.model_dump() # defaults to by_alias=True - >>> # For compact, deterministic output: - >>> schema = cv.io_to_invest(include_audit_log=False) - >>> assert schema.audit_log is None """ - return serialize_investigation(self._investigation, include_audit_log=include_audit_log) + return serialize_investigation(self._investigation) - def io_to_dict(self, *, include_audit_log: bool = True) -> dict[str, Any]: + def io_to_dict(self) -> dict[str, Any]: """ Convert the investigation to a Python dictionary. - Args: - include_audit_log: Include audit log in output (default: True). - When False, audit_log is set to None for compact, deterministic output. - Returns: Dictionary representation of the investigation @@ -1334,11 +1315,8 @@ def io_to_dict(self, *, include_audit_log: bool = True) -> dict[str, Any]: >>> cv = Cyvest() >>> data = cv.io_to_dict() >>> print(data["score"], data["level"]) - >>> # For compact, deterministic output: - >>> data = cv.io_to_dict(include_audit_log=False) - >>> assert data["audit_log"] is None """ - return self.io_to_invest(include_audit_log=include_audit_log).model_dump(by_alias=True) + return self.io_to_invest().model_dump(by_alias=True) @staticmethod def io_load_json(filepath: str | Path) -> Cyvest: @@ -1534,7 +1512,6 @@ def display_summary( self, show_graph: bool = True, exclude_levels: Level | Iterable[Level] = Level.NONE, - show_audit_log: bool = False, rich_print: Callable[[Any], None] | None = None, ) -> None: """ @@ -1543,7 +1520,6 @@ def display_summary( Args: show_graph: Whether to display the observable graph exclude_levels: Level(s) to omit from the report (default: Level.NONE) - show_audit_log: Whether to display the investigation audit log rich_print: Optional callable that takes a renderable and returns None """ if rich_print is None: @@ -1556,7 +1532,6 @@ def rich_print(renderables: Any) -> None: rich_print, show_graph=show_graph, exclude_levels=exclude_levels, - show_audit_log=show_audit_log, ) def display_statistics( diff --git a/src/cyvest/investigation.py b/src/cyvest/investigation.py index b2c51f6..14f737d 100644 --- a/src/cyvest/investigation.py +++ b/src/cyvest/investigation.py @@ -18,7 +18,6 @@ from cyvest.level_score_rules import recalculate_level_for_score from cyvest.levels import Level, normalize_level from cyvest.model import ( - AuditEvent, Enrichment, Evidence, EvidenceLink, @@ -108,15 +107,7 @@ def __init__( """ self.investigation_id = investigation_id or generate_ulid() self.investigation_name = investigation_name - self._audit_log: list[AuditEvent] = [] - self._audit_enabled = True - - # Record investigation start as the first event - self._record_event( - event_type="INVESTIGATION_STARTED", - object_type="investigation", - object_key=self.investigation_id, - ) + self._started_at = datetime.now(timezone.utc) # Object collections self._observables: dict[str, Observable] = {} @@ -148,49 +139,11 @@ def __init__( self._observables[self._root_observable.key] = self._root_observable self._score_engine.register_observable(self._root_observable) self._stats.register_observable(self._root_observable) - self._record_event( - event_type="OBSERVABLE_CREATED", - object_type="observable", - object_key=self._root_observable.key, - ) - - def _record_event( - self, - *, - event_type: str, - object_type: str | None = None, - object_key: str | None = None, - reason: str | None = None, - actor: str | None = None, - tool: str | None = None, - details: dict[str, Any] | None = None, - timestamp: datetime | None = None, - ) -> AuditEvent | None: - if not self._audit_enabled: - return None - - event = AuditEvent( - event_id=generate_ulid(), - timestamp=timestamp or datetime.now(timezone.utc), - event_type=event_type, - actor=actor, - reason=reason, - tool=tool, - object_type=object_type, - object_key=object_key, - details=deepcopy(details) if details else {}, - ) - self._audit_log.append(event) - return event @property def started_at(self) -> datetime: - """Return the investigation start time from the first event in the audit log.""" - for event in self._audit_log: - if event.event_type == "INVESTIGATION_STARTED": - return event.timestamp - # Fallback if no INVESTIGATION_STARTED event (shouldn't happen) - return datetime.now(timezone.utc) + """Return the investigation start time.""" + return self._started_at def _link_threat_intel_to_observable(self, observable: Observable, ti: ThreatIntel) -> None: if any(existing.key == ti.key for existing in observable.threat_intels): @@ -232,21 +185,6 @@ def _link_finding_to_tag(self, tag: Tag, finding: Finding) -> None: return tag.findings.append(finding) - def _get_object_type(self, obj: Any) -> str | None: - if isinstance(obj, Observable): - return "observable" - if isinstance(obj, Finding): - return "finding" - if isinstance(obj, Evidence): - return "evidence" - if isinstance(obj, ThreatIntel): - return "threat_intel" - if isinstance(obj, Enrichment): - return "enrichment" - if isinstance(obj, Tag): - return "tag" - return None - @staticmethod def _normalize_taxonomies(value: Any) -> list[Taxonomy]: if value is None: @@ -274,7 +212,7 @@ def apply_score_change( event_type: str = "SCORE_CHANGED", contributing_investigation_ids: set[str] | None = None, ) -> bool: - """Apply a score change and emit an audit event.""" + """Apply a score change.""" if not isinstance(new_score, Decimal): new_score = Decimal(str(new_score)) @@ -287,27 +225,6 @@ def apply_score_change( obj.score = new_score obj.level = new_level - - if event_type == "SCORE_RECALCULATED": - # Skip audit log entry for recalculated scores. - return True - - details = { - "old_score": float(old_score), - "new_score": float(new_score), - "old_level": old_level.value, - "new_level": new_level.value, - } - if contributing_investigation_ids: - details["contributing_investigation_ids"] = sorted(contributing_investigation_ids) - - self._record_event( - event_type=event_type, - object_type=self._get_object_type(obj), - object_key=getattr(obj, "key", None), - reason=reason, - details=details, - ) return True def apply_level_change( @@ -318,24 +235,13 @@ def apply_level_change( reason: str = "", event_type: str = "LEVEL_UPDATED", ) -> bool: - """Apply a level change and emit an audit event.""" + """Apply a level change.""" new_level = normalize_level(level) old_level = obj.level if new_level == old_level: return False obj.level = new_level - self._record_event( - event_type=event_type, - object_type=self._get_object_type(obj), - object_key=getattr(obj, "key", None), - reason=reason, - details={ - "old_level": old_level.value, - "new_level": new_level.value, - "score": float(obj.score), - }, - ) return True def _update_observable_finding_links(self, observable_key: str) -> None: @@ -363,41 +269,12 @@ def _rebuild_all_evidence_finding_links(self) -> None: for evidence_key in self._evidences: self._update_evidence_finding_links(evidence_key) - def get_audit_log(self) -> list[AuditEvent]: - """Return a deep copy of the audit log.""" - return [event.model_copy(deep=True) for event in self._audit_log] - - def get_audit_events( - self, - *, - object_type: str | None = None, - object_key: str | None = None, - event_type: str | None = None, - ) -> list[AuditEvent]: - """Filter audit events by optional object type/key and event type.""" - events = self._audit_log - if object_type is not None: - events = [event for event in events if event.object_type == object_type] - if object_key is not None: - events = [event for event in events if event.object_key == object_key] - if event_type is not None: - events = [event for event in events if event.event_type == event_type] - return [event.model_copy(deep=True) for event in events] - def set_investigation_name(self, name: str | None, *, reason: str | None = None) -> None: """Set or clear the human-readable investigation name.""" name = str(name).strip() if name is not None else None if name == self.investigation_name: return - old_name = self.investigation_name self.investigation_name = name - self._record_event( - event_type="INVESTIGATION_NAME_UPDATED", - object_type="investigation", - object_key=self.investigation_id, - reason=reason, - details={"old_name": old_name, "new_name": name}, - ) def _merge_observable(self, existing: Observable, incoming: Observable) -> tuple[Observable, list]: """ @@ -758,11 +635,6 @@ def add_observable(self, obs: Observable) -> tuple[Observable, list]: self._score_engine.register_observable(obs) self._stats.register_observable(obs) self._update_observable_finding_links(obs.key) - self._record_event( - event_type="OBSERVABLE_CREATED", - object_type="observable", - object_key=obs.key, - ) return obs, [] def add_finding(self, finding: Finding) -> Finding: @@ -796,11 +668,6 @@ def add_finding(self, finding: Finding) -> Finding: self._update_observable_finding_links(link.observable_key) for link in finding.evidence_links: self._update_evidence_finding_links(link.evidence_key) - self._record_event( - event_type="FINDING_CREATED", - object_type="finding", - object_key=finding.key, - ) return finding def add_evidence(self, evidence: Evidence) -> Evidence: @@ -810,11 +677,6 @@ def add_evidence(self, evidence: Evidence) -> Evidence: self._evidences[evidence.key] = evidence self._stats.register_evidence(evidence) self._update_evidence_finding_links(evidence.key) - self._record_event( - event_type="EVIDENCE_CREATED", - object_type="evidence", - object_key=evidence.key, - ) return evidence def add_threat_intel(self, ti: ThreatIntel, observable: Observable) -> ThreatIntel: @@ -832,17 +694,6 @@ def add_threat_intel(self, ti: ThreatIntel, observable: Observable) -> ThreatInt merged_ti = self._merge_threat_intel(self._threat_intels[ti.key], ti) # Propagate score to observable self._score_engine.propagate_threat_intel_to_observable(merged_ti, observable) - self._record_event( - event_type="THREAT_INTEL_ATTACHED", - object_type="observable", - object_key=observable.key, - details={ - "threat_intel_key": merged_ti.key, - "source": merged_ti.source, - "score": merged_ti.score, - "level": merged_ti.level, - }, - ) return merged_ti # Register new threat intel @@ -855,17 +706,6 @@ def add_threat_intel(self, ti: ThreatIntel, observable: Observable) -> ThreatInt # Propagate score self._score_engine.propagate_threat_intel_to_observable(ti, observable) - self._record_event( - event_type="THREAT_INTEL_ATTACHED", - object_type="observable", - object_key=observable.key, - details={ - "threat_intel_key": ti.key, - "source": ti.source, - "score": ti.score, - "level": ti.level, - }, - ) return ti def add_threat_intel_taxonomy(self, threat_intel_key: str, taxonomy: Taxonomy) -> ThreatIntel: @@ -932,11 +772,6 @@ def add_enrichment(self, enrichment: Enrichment) -> Enrichment: # Register new enrichment self._enrichments[enrichment.key] = enrichment - self._record_event( - event_type="ENRICHMENT_CREATED", - object_type="enrichment", - object_key=enrichment.key, - ) return enrichment def add_tag(self, tag: Tag) -> Tag: @@ -962,12 +797,6 @@ def add_tag(self, tag: Tag) -> Tag: ancestor_tag = Tag(name=ancestor_name) self._tags[ancestor_key] = ancestor_tag self._stats.register_tag(ancestor_tag) - self._record_event( - event_type="TAG_CREATED", - object_type="tag", - object_key=ancestor_key, - details={"auto_created": True, "descendant": tag.name}, - ) # Add or merge the tag itself if tag.key in self._tags: @@ -978,11 +807,6 @@ def add_tag(self, tag: Tag) -> Tag: # Register new tag self._tags[tag.key] = tag self._stats.register_tag(tag) - self._record_event( - event_type="TAG_CREATED", - object_type="tag", - object_key=tag.key, - ) return tag def add_relationship( @@ -1045,17 +869,6 @@ def add_relationship( # Add relationship using internal method self._create_relationship(source_obs, target_key, relationship_type, direction) - self._record_event( - event_type="RELATIONSHIP_CREATED", - object_type="observable", - object_key=source_obs.key, - details={ - "target_key": target_key, - "relationship_type": relationship_type, - "direction": direction, - }, - ) - # Recalculate scores after adding relationship self._score_engine.recalculate_all() @@ -1102,15 +915,6 @@ def link_finding_observable( observable_key=observable_key, ) self._update_observable_finding_links(observable_key) - self._record_event( - event_type="FINDING_LINKED_TO_OBSERVABLE", - object_type="finding", - object_key=finding.key, - details={ - "observable_key": observable_key, - "propagation_mode": propagation_mode.value, - }, - ) is_effective = ( propagation_mode == PropagationMode.GLOBAL or self.investigation_id == finding.origin_investigation_id @@ -1133,12 +937,6 @@ def link_finding_evidence(self, finding_key: str, evidence_key: str) -> Finding: if self._link_finding_to_evidence(finding, EvidenceLink(evidence_key=evidence_key)): self._update_evidence_finding_links(evidence_key) - self._record_event( - event_type="FINDING_LINKED_TO_EVIDENCE", - object_type="finding", - object_key=finding.key, - details={"evidence_key": evidence_key}, - ) return finding def add_finding_to_tag(self, tag_key: str, finding_key: str) -> Tag: @@ -1165,12 +963,6 @@ def add_finding_to_tag(self, tag_key: str, finding_key: str) -> Tag: if tag and finding: self._link_finding_to_tag(tag, finding) - self._record_event( - event_type="TAG_FINDING_ADDED", - object_type="tag", - object_key=tag.key, - details={"finding_key": finding.key}, - ) return tag @@ -1326,14 +1118,11 @@ def update_model_metadata( allowed_fields = rules["fields"] dict_fields = rules["dict_fields"] - changes: dict[str, dict[str, Any]] = {} - for field, value in updates.items(): if field not in allowed_fields: raise ValueError(f"Field '{field}' is not mutable on {model_type}.") if value is None: continue - old_value = deepcopy(getattr(target, field, None)) if field == "level": value = normalize_level(value) if model_type == "threat_intel" and field == "taxonomies": @@ -1356,17 +1145,7 @@ def update_model_metadata( setattr(target, field, deepcopy(value)) else: setattr(target, field, value) - new_value = deepcopy(getattr(target, field, None)) - if old_value != new_value: - changes[field] = {"old": old_value, "new": new_value} - - if changes: - self._record_event( - event_type="METADATA_UPDATED", - object_type=model_type, - object_key=key, - details={"changes": changes}, - ) + return target def get_all_observables(self) -> dict[str, Observable]: @@ -1428,16 +1207,6 @@ def add_whitelist(self, identifier: str, name: str, justification: str | None = entry = InvestigationWhitelist(identifier=identifier, name=name, justification=justification) self._whitelists[identifier] = entry - self._record_event( - event_type="WHITELIST_APPLIED", - object_type="investigation", - object_key=self.investigation_id, - details={ - "identifier": identifier, - "name": name, - "justification": justification, - }, - ) return entry def remove_whitelist(self, identifier: str) -> bool: @@ -1448,27 +1217,11 @@ def remove_whitelist(self, identifier: str) -> bool: True if removed, False if it did not exist. """ removed = self._whitelists.pop(identifier, None) - if removed: - self._record_event( - event_type="WHITELIST_REMOVED", - object_type="investigation", - object_key=self.investigation_id, - details={"identifier": identifier}, - ) return removed is not None def clear_whitelists(self) -> None: """Remove all whitelist entries.""" - if not self._whitelists: - return - removed = list(self._whitelists.keys()) self._whitelists.clear() - self._record_event( - event_type="WHITELIST_CLEARED", - object_type="investigation", - object_key=self.investigation_id, - details={"identifiers": removed}, - ) def get_whitelists(self) -> list[InvestigationWhitelist]: """Return a copy of all whitelist entries.""" @@ -1549,17 +1302,6 @@ def bfs(start_key: str) -> set[str]: # Link the best starting node to root if best_node: self._create_relationship(self._root_observable, best_node, RelationshipType.RELATED_TO) - self._record_event( - event_type="RELATIONSHIP_CREATED", - object_type="observable", - object_key=self._root_observable.key, - reason="Finalize relationships", - details={ - "target_key": best_node, - "relationship_type": RelationshipType.RELATED_TO.value, - "direction": RelationshipType.RELATED_TO.get_default_direction().value, - }, - ) self._score_engine.recalculate_all() def merge_investigation(self, other: Investigation) -> None: @@ -1573,80 +1315,6 @@ def merge_investigation(self, other: Investigation) -> None: Args: other: The investigation to merge """ - - def _diff_fields(before: dict[str, Any], after: dict[str, Any]) -> list[str]: - return [field for field, value in before.items() if value != after.get(field)] - - def _snapshot_observable(obs: Observable) -> dict[str, Any]: - relationships = [ - ( - rel.target_key, - rel.relationship_type_name, - rel.direction.value, - ) - for rel in obs.relationships - ] - return { - "score": obs.score, - "level": obs.level, - "comment": obs.comment, - "extra": deepcopy(obs.extra), - "internal": obs.internal, - "whitelisted": obs.whitelisted, - "threat_intels": sorted(ti.key for ti in obs.threat_intels), - "relationships": sorted(relationships), - } - - def _snapshot_finding(finding: Finding) -> dict[str, Any]: - links = [ - ( - link.observable_key, - link.propagation_mode.value, - ) - for link in finding.observable_links - ] - return { - "score": finding.score, - "level": finding.level, - "comment": finding.comment, - "description": finding.description, - "extra": deepcopy(finding.extra), - "origin_investigation_id": finding.origin_investigation_id, - "observable_links": sorted(links), - "evidence_links": sorted(link.evidence_key for link in finding.evidence_links), - } - - def _snapshot_evidence(evidence: Evidence) -> dict[str, Any]: - return { - "title": evidence.title, - "description": evidence.description, - "extra": deepcopy(evidence.extra), - "captured_at": evidence.captured_at, - } - - def _snapshot_threat_intel(ti: ThreatIntel) -> dict[str, Any]: - return { - "score": ti.score, - "level": ti.level, - "comment": ti.comment, - "extra": deepcopy(ti.extra), - "taxonomies": deepcopy(ti.taxonomies), - } - - def _snapshot_enrichment(enrichment: Enrichment) -> dict[str, Any]: - return { - "context": enrichment.context, - "data": deepcopy(enrichment.data), - } - - def _snapshot_tag(tag: Tag) -> dict[str, Any]: - return { - "description": tag.description, - "findings": sorted(finding.key for finding in tag.findings), - } - - merge_summary: list[dict[str, Any]] = [] - ( incoming_observables, incoming_threat_intels, @@ -1659,31 +1327,8 @@ def _snapshot_tag(tag: Tag) -> dict[str, Any]: # PASS 1: Merge observables and collect deferred relationships all_deferred_relationships = [] for obs in incoming_observables.values(): - existing = self._observables.get(obs.key) - before = _snapshot_observable(existing) if existing else None _, deferred = self.add_observable(obs) all_deferred_relationships.extend(deferred) - if existing: - after = _snapshot_observable(existing) - changed_fields = _diff_fields(before, after) if before else [] - action = "merged" if changed_fields else "skipped" - merge_summary.append( - { - "object_type": "observable", - "object_key": obs.key, - "action": action, - "changed_fields": changed_fields, - } - ) - else: - merge_summary.append( - { - "object_type": "observable", - "object_key": obs.key, - "action": "created", - "changed_fields": [], - } - ) # PASS 2: Process deferred relationships now that all observables exist for source_key, rel in all_deferred_relationships: @@ -1702,111 +1347,26 @@ def _snapshot_tag(tag: Tag) -> dict[str, Any]: # Merge threat intels (need to link to observables) for ti in incoming_threat_intels.values(): - existing_ti = self._threat_intels.get(ti.key) - before = _snapshot_threat_intel(existing_ti) if existing_ti else None # Find the observable this TI belongs to observable = self._observables.get(ti.observable_key) if observable: self.add_threat_intel(ti, observable) - if existing_ti: - after = _snapshot_threat_intel(existing_ti) - changed_fields = _diff_fields(before, after) if before else [] - action = "merged" if changed_fields else "skipped" - else: - changed_fields = [] - action = "created" - merge_summary.append( - { - "object_type": "threat_intel", - "object_key": ti.key, - "action": action, - "changed_fields": changed_fields, - } - ) # Merge evidences before findings so all links have valid targets. for evidence in incoming_evidences.values(): - existing_evidence = self._evidences.get(evidence.key) - before = _snapshot_evidence(existing_evidence) if existing_evidence else None self.add_evidence(evidence) - if existing_evidence: - after = _snapshot_evidence(existing_evidence) - changed_fields = _diff_fields(before, after) if before else [] - action = "merged" if changed_fields else "skipped" - else: - changed_fields = [] - action = "created" - merge_summary.append( - { - "object_type": "evidence", - "object_key": evidence.key, - "action": action, - "changed_fields": changed_fields, - } - ) # Merge findings for finding in incoming_findings.values(): - existing_finding = self._findings.get(finding.key) - before = _snapshot_finding(existing_finding) if existing_finding else None self.add_finding(finding) - if existing_finding: - after = _snapshot_finding(existing_finding) - changed_fields = _diff_fields(before, after) if before else [] - action = "merged" if changed_fields else "skipped" - else: - changed_fields = [] - action = "created" - merge_summary.append( - { - "object_type": "finding", - "object_key": finding.key, - "action": action, - "changed_fields": changed_fields, - } - ) # Merge enrichments for enrichment in incoming_enrichments.values(): - existing_enrichment = self._enrichments.get(enrichment.key) - before = _snapshot_enrichment(existing_enrichment) if existing_enrichment else None self.add_enrichment(enrichment) - if existing_enrichment: - after = _snapshot_enrichment(existing_enrichment) - changed_fields = _diff_fields(before, after) if before else [] - action = "merged" if changed_fields else "skipped" - else: - changed_fields = [] - action = "created" - merge_summary.append( - { - "object_type": "enrichment", - "object_key": enrichment.key, - "action": action, - "changed_fields": changed_fields, - } - ) # Merge tags for tag in incoming_tags.values(): - existing_tag = self._tags.get(tag.key) - before = _snapshot_tag(existing_tag) if existing_tag else None self.add_tag(tag) - if existing_tag: - after = _snapshot_tag(existing_tag) - changed_fields = _diff_fields(before, after) if before else [] - action = "merged" if changed_fields else "skipped" - else: - changed_fields = [] - action = "created" - merge_summary.append( - { - "object_type": "tag", - "object_key": tag.key, - "action": action, - "changed_fields": changed_fields, - } - ) # Merge whitelists (other investigation overrides on identifier conflicts) for entry in other.get_whitelists(): @@ -1819,16 +1379,3 @@ def _snapshot_tag(tag: Tag) -> dict[str, Any]: # Final score recalculation self._score_engine.recalculate_all() - - self._record_event( - event_type="INVESTIGATION_MERGED", - object_type="investigation", - object_key=self.investigation_id, - details={ - "from_investigation_id": other.investigation_id, - "into_investigation_id": self.investigation_id, - "from_investigation_name": other.investigation_name, - "into_investigation_name": self.investigation_name, - "object_changes": merge_summary, - }, - ) diff --git a/src/cyvest/io_rich.py b/src/cyvest/io_rich.py index 45eb64f..b170dfd 100644 --- a/src/cyvest/io_rich.py +++ b/src/cyvest/io_rich.py @@ -8,7 +8,6 @@ import json from collections.abc import Callable, Iterable -from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from typing import TYPE_CHECKING, Any @@ -140,205 +139,11 @@ def _build_observable_tree( ) -def _render_audit_log_table( - *, - rich_print: Callable[[Any], None], - title: str, - events: Iterable[Any], - started_at: datetime | None, -) -> None: - def _render_score_change(details: dict[str, Any]) -> str: - old_score = details.get("old_score") - new_score = details.get("new_score") - old_level = details.get("old_level") - new_level = details.get("new_level") - - parts: list[str] = [] - if old_score is not None and new_score is not None: - old_score = old_score if isinstance(old_score, Decimal) else Decimal(str(old_score)) - new_score = new_score if isinstance(new_score, Decimal) else Decimal(str(new_score)) - old_score_color = get_color_score(old_score) - new_score_color = get_color_score(new_score) - score_str = ( - f"[{old_score_color}]{_format_score_decimal(old_score)}[/" - f"{old_score_color}] → " - f"[{new_score_color}]{_format_score_decimal(new_score)}[/" - f"{new_score_color}]" - ) - parts.append(f"Score: {score_str}") - - if old_level is not None and new_level is not None: - old_level_enum = normalize_level(old_level) - new_level_enum = normalize_level(new_level) - old_level_color = get_color_level(old_level_enum) - new_level_color = get_color_level(new_level_enum) - level_str = ( - f"[{old_level_color}]{old_level_enum.name}[/" - f"{old_level_color}] → " - f"[{new_level_color}]{new_level_enum.name}[/" - f"{new_level_color}]" - ) - parts.append(f"Level: {level_str}") - - return " | ".join(parts) if parts else "[dim]-[/dim]" - - def _render_level_change(details: dict[str, Any]) -> str: - old_level = details.get("old_level") - new_level = details.get("new_level") - score = details.get("score") - if old_level is None or new_level is None: - return "[dim]-[/dim]" - old_level_enum = normalize_level(old_level) - new_level_enum = normalize_level(new_level) - old_level_color = get_color_level(old_level_enum) - new_level_color = get_color_level(new_level_enum) - level_str = ( - f"[{old_level_color}]{old_level_enum.name}[/" - f"{old_level_color}] → " - f"[{new_level_color}]{new_level_enum.name}[/" - f"{new_level_color}]" - ) - if score is None: - return f"Level: {level_str}" - score = score if isinstance(score, Decimal) else Decimal(str(score)) - score_color = get_color_score(score) - score_str = f"[{score_color}]{_format_score_decimal(score)}[/{score_color}]" - return f"Level: {level_str} | Score: {score_str}" - - def _render_merge_event(details: dict[str, Any]) -> str: - from_name = details.get("from_investigation_name") - into_name = details.get("into_investigation_name") - from_id = details.get("from_investigation_id") - into_id = details.get("into_investigation_id") - from_label = escape(str(from_name)) if from_name else escape(str(from_id)) - into_label = escape(str(into_name)) if into_name else escape(str(into_id)) - if not from_label or from_label == "None": - from_label = "[dim]-[/dim]" - if not into_label or into_label == "None": - into_label = "[dim]-[/dim]" - - object_changes = details.get("object_changes") or [] - counts: dict[str, int] = {} - for change in object_changes: - action = change.get("action") - if not action: - continue - counts[action] = counts.get(action, 0) + 1 - - if counts: - parts = [f"{key}={value}" for key, value in sorted(counts.items())] - summary = ", ".join(parts) - return f"Merge: {from_label} → {into_label} | Changes: {summary}" - - return f"Merge: {from_label} → {into_label}" - - def _render_threat_intel_attached(details: dict[str, Any]) -> str: - source = details.get("source") - score = details.get("score") - level = details.get("level") - parts: list[str] = [] - if source: - parts.append(f"Source: [cyan]{escape(str(source))}[/cyan]") - if level is not None: - level_enum = normalize_level(level) - level_color = get_color_level(level_enum) - parts.append(f"Level: [{level_color}]{level_enum.name}[/{level_color}]") - if score is not None: - score_value = score if isinstance(score, Decimal) else Decimal(str(score)) - score_color = get_color_score(score_value) - score_str = f"[{score_color}]{_format_score_decimal(score_value)}[/{score_color}]" - parts.append(f"Score: {score_str}") - return " | ".join(parts) if parts else "[dim]-[/dim]" - - detail_renderers: dict[str, Callable[[dict[str, Any]], str]] = { - "SCORE_CHANGED": _render_score_change, - "SCORE_RECALCULATED": _render_score_change, - "LEVEL_UPDATED": _render_level_change, - "INVESTIGATION_MERGED": _render_merge_event, - "THREAT_INTEL_ATTACHED": _render_threat_intel_attached, - } - - def _coerce_utc(value: datetime) -> datetime: - if value.tzinfo is None: - return value.replace(tzinfo=timezone.utc) - return value.astimezone(timezone.utc) - - def _format_elapsed(total_seconds: float) -> str: - total_ms = int(round(total_seconds * 1000)) - if total_ms < 0: - total_ms = 0 - hours, rem_ms = divmod(total_ms, 3_600_000) - minutes, rem_ms = divmod(rem_ms, 60_000) - seconds, ms = divmod(rem_ms, 1000) - return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{ms:03d}" - - table = Table(title=title, show_lines=False) - table.add_column("#", justify="right") - table.add_column("Elapsed", style="dim") - table.add_column("Event") - table.add_column("Object") - table.add_column("Context") - - events_sorted = sorted(events, key=lambda evt: evt.timestamp) - effective_start = _coerce_utc(started_at) if started_at is not None else None - if effective_start is None and events_sorted: - effective_start = _coerce_utc(events_sorted[0].timestamp) - - grouped_events: dict[str, list[Any]] = {} - group_order: list[str] = [] - for event in events_sorted: - group_key = event.object_key or "" - if group_key not in grouped_events: - grouped_events[group_key] = [] - group_order.append(group_key) - grouped_events[group_key].append(event) - - row_idx = 1 - for group_key in group_order: - if row_idx > 1: - table.add_section() - for event in grouped_events[group_key]: - event_timestamp = _coerce_utc(event.timestamp) - elapsed = "" - if effective_start is not None: - elapsed = _format_elapsed((event_timestamp - effective_start).total_seconds()) - - event_type = escape(event.event_type) - object_label = "[dim]-[/dim]" - if event.object_key: - object_label = escape(event.object_key) - reason = escape(event.reason) if event.reason else "[dim]-[/dim]" - details = "[dim]-[/dim]" - renderer = detail_renderers.get(event.event_type) - if renderer: - details = renderer(getattr(event, "details", {}) or {}) - - if reason == "[dim]-[/dim]": - context = details - elif details == "[dim]-[/dim]": - context = reason - else: - context = details - - table.add_row( - str(row_idx), - elapsed, - event_type, - object_label, - context, - ) - row_idx += 1 - - table.caption = "No audit events recorded." if not events_sorted else "" - rich_print(table) - - def display_summary( cv: Cyvest, rich_print: Callable[[Any], None], show_graph: bool = True, exclude_levels: Level | Iterable[Level] = Level.NONE, - show_audit_log: bool = False, ) -> None: """ Display a comprehensive summary of the investigation using Rich. @@ -348,7 +153,6 @@ def display_summary( rich_print: A rich renderable handler that is called with renderables for output show_graph: Whether to display the observable graph exclude_levels: Level(s) to omit from the report (default: Level.NONE) - show_audit_log: Whether to display the investigation audit log (default: False) """ resolved_excluded_levels = _normalize_exclude_levels(exclude_levels) @@ -485,18 +289,6 @@ def display_summary( rich_print(tree) - if show_audit_log: - investigation = getattr(cv, "_investigation", None) - events = investigation.get_audit_log() if investigation else [] - if events: - started_at = investigation.started_at if investigation else None - _render_audit_log_table( - rich_print=rich_print, - title="Audit Log", - events=events, - started_at=started_at, - ) - def display_statistics(cv: Cyvest, rich_print: Callable[[Any], None]) -> None: """ diff --git a/src/cyvest/io_serialization.py b/src/cyvest/io_serialization.py index 4eac025..0ab8fc9 100644 --- a/src/cyvest/io_serialization.py +++ b/src/cyvest/io_serialization.py @@ -14,7 +14,7 @@ from cyvest import keys from cyvest.levels import Level, normalize_level -from cyvest.model import AuditEvent, Enrichment, Evidence, Finding, Observable, Relationship, Tag, ThreatIntel +from cyvest.model import Enrichment, Evidence, Finding, Observable, Relationship, Tag, ThreatIntel from cyvest.model_enums import ObservableType from cyvest.model_schema import InvestigationSchema from cyvest.score import ScoreMode @@ -24,7 +24,7 @@ from cyvest.investigation import Investigation -def serialize_investigation(inv: Investigation, *, include_audit_log: bool = True) -> InvestigationSchema: +def serialize_investigation(inv: Investigation) -> InvestigationSchema: """ Serialize a complete investigation to an InvestigationSchema. @@ -33,8 +33,6 @@ def serialize_investigation(inv: Investigation, *, include_audit_log: bool = Tru Args: inv: Investigation to serialize - include_audit_log: Include audit log in serialization (default: True). - When False, audit_log is set to None for compact, deterministic output. Returns: InvestigationSchema instance (use .model_dump() for dict) @@ -62,7 +60,6 @@ def serialize_investigation(inv: Investigation, *, include_audit_log: bool = Tru level=inv.get_global_level(), whitelisted=inv.is_whitelisted(), whitelists=list(inv.get_whitelists()), - audit_log=inv.get_audit_log() if include_audit_log else None, observables=observables, findings=findings, evidences=evidences, @@ -79,17 +76,15 @@ def serialize_investigation(inv: Investigation, *, include_audit_log: bool = Tru return investigation -def save_investigation_json(inv: Investigation, filepath: str | Path, *, include_audit_log: bool = True) -> None: +def save_investigation_json(inv: Investigation, filepath: str | Path) -> None: """ Save an investigation to a JSON file. Args: inv: Investigation to save filepath: Path to save the JSON file - include_audit_log: Include audit log in output (default: True). - When False, audit_log is set to null for compact, deterministic output. """ - data = serialize_investigation(inv, include_audit_log=include_audit_log) + data = serialize_investigation(inv) with open(filepath, "w", encoding="utf-8") as f: f.write(data.model_dump_json(indent=2, by_alias=True)) @@ -308,8 +303,6 @@ def load_investigation_dict(data: dict[str, Any]) -> Cyvest: score_mode_obs=score_mode, investigation_id=investigation_id, ) - cv._investigation._audit_enabled = False - cv._investigation._audit_log = [] investigation_name = data.get("investigation_name") if isinstance(investigation_name, str): @@ -481,15 +474,6 @@ def build_tag(tag_info: dict[str, Any]) -> Tag: cv._investigation._rebuild_all_finding_links() cv._investigation._rebuild_all_evidence_finding_links() - audit_log = [] - for event_info in data.get("audit_log", []) or []: - try: - audit_log.append(AuditEvent.model_validate(event_info)) - except Exception: - continue - cv._investigation._audit_log = audit_log - cv._investigation._audit_enabled = True - return cv @@ -568,29 +552,6 @@ def migrate_v5_to_v6(data: dict[str, Any]) -> dict[str, Any]: old_finding_keys = tag.pop("checks", tag.get("findings", [])) tag["findings"] = [finding_key_map.get(key, key) for key in old_finding_keys] - reference_map = {**observable_key_map, **threat_intel_key_map, **finding_key_map} - - def rewrite_audit_value(value: Any) -> Any: - if isinstance(value, str): - return reference_map.get(value, value) - if isinstance(value, list): - return [rewrite_audit_value(item) for item in value] - if isinstance(value, dict): - rewritten: dict[str, Any] = {} - for key, item in value.items(): - rewritten_key = key.replace("check", "finding") - rewritten[rewritten_key] = rewrite_audit_value(item) - return rewritten - return value - - for event in migrated.get("audit_log", []) or []: - if event.get("object_type") == "check": - event["object_type"] = "finding" - if isinstance(event.get("event_type"), str): - event["event_type"] = event["event_type"].replace("CHECK", "FINDING") - event["object_key"] = reference_map.get(event.get("object_key"), event.get("object_key")) - event["details"] = rewrite_audit_value(event.get("details", {})) - migrated.pop("stats", None) loaded = load_investigation_dict(migrated) return serialize_investigation(loaded._investigation).model_dump(mode="json", by_alias=True) diff --git a/src/cyvest/model.py b/src/cyvest/model.py index adf5ff7..c0b5b73 100644 --- a/src/cyvest/model.py +++ b/src/cyvest/model.py @@ -71,22 +71,6 @@ def _format_score_decimal(value: Decimal | None, *, places: int = _DEFAULT_SCORE return str(value) -class AuditEvent(BaseModel): - """Centralized audit event for investigation-level changes.""" - - model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow") - - event_id: str - timestamp: datetime - event_type: str - actor: str | None = None - reason: str | None = None - tool: str | None = None - object_type: str | None = None - object_key: str | None = None - details: dict[str, Any] = Field(default_factory=dict) - - class InvestigationWhitelist(BaseModel): """Represents a whitelist entry on an investigation.""" diff --git a/src/cyvest/model_schema.py b/src/cyvest/model_schema.py index e2525c4..fa0c0fa 100644 --- a/src/cyvest/model_schema.py +++ b/src/cyvest/model_schema.py @@ -20,7 +20,6 @@ from cyvest.levels import Level from cyvest.model import ( AliasDumpModel, - AuditEvent, Enrichment, Evidence, Finding, @@ -114,10 +113,6 @@ class InvestigationSchema(AliasDumpModel): ..., description="List of whitelist entries applied to this investigation.", ) - audit_log: list[AuditEvent] | None = Field( - default_factory=list, - description="Append-only investigation audit log. Null when serialization disabled audit.", - ) observables: dict[str, Observable] = Field( ..., description="Observables keyed by their unique key.", @@ -163,7 +158,6 @@ def ensure_defaults(cls, v: Any) -> Any: v.setdefault("level", Level.NONE) v.setdefault("whitelists", []) - v.setdefault("audit_log", []) v.setdefault("observables", {}) v.setdefault("findings", {}) v.setdefault("evidences", {}) diff --git a/src/cyvest/proxies.py b/src/cyvest/proxies.py index ac4b999..a64a164 100644 --- a/src/cyvest/proxies.py +++ b/src/cyvest/proxies.py @@ -167,11 +167,6 @@ def finding_links(self) -> list[str]: """Findings that currently link to this observable.""" return self._read_attr("finding_links") - def get_audit_events(self) -> tuple: - """Return audit events for this observable.""" - events = self._get_investigation().get_audit_events(object_type="observable", object_key=self.key) - return tuple(events) - def update_metadata( self, *, @@ -347,11 +342,6 @@ def observable_links(self) -> list[ObservableLink]: def evidence_links(self) -> list[EvidenceLink]: return self._read_attr("evidence_links") - def get_audit_events(self) -> tuple: - """Return audit events for this finding.""" - events = self._get_investigation().get_audit_events(object_type="finding", object_key=self.key) - return tuple(events) - def update_metadata( self, *, diff --git a/src/cyvest/shared.py b/src/cyvest/shared.py index ed5efc4..1ceacba 100644 --- a/src/cyvest/shared.py +++ b/src/cyvest/shared.py @@ -531,21 +531,21 @@ def _io_save_markdown_unlocked( ) return str(Path(filepath).resolve()) - def io_to_invest(self, *, include_audit_log: bool = True) -> InvestigationSchema: - return self._lock.run(self._io_to_invest_unlocked, include_audit_log) + def io_to_invest(self) -> InvestigationSchema: + return self._lock.run(self._io_to_invest_unlocked) - async def aio_to_invest(self, *, include_audit_log: bool = True) -> InvestigationSchema: - return await self._lock.arun(self._io_to_invest_unlocked, include_audit_log) + async def aio_to_invest(self) -> InvestigationSchema: + return await self._lock.arun(self._io_to_invest_unlocked) - def _io_to_invest_unlocked(self, include_audit_log: bool = True) -> InvestigationSchema: - return serialize_investigation(self._main_investigation, include_audit_log=include_audit_log) + def _io_to_invest_unlocked(self) -> InvestigationSchema: + return serialize_investigation(self._main_investigation) - def io_save_json(self, filepath: str | Path, *, include_audit_log: bool = True) -> str: - return self._lock.run(self._io_save_json_unlocked, filepath, include_audit_log) + def io_save_json(self, filepath: str | Path) -> str: + return self._lock.run(self._io_save_json_unlocked, filepath) - async def aio_save_json(self, filepath: str | Path, *, include_audit_log: bool = True) -> str: - return await self._lock.arun(self._io_save_json_unlocked, filepath, include_audit_log) + async def aio_save_json(self, filepath: str | Path) -> str: + return await self._lock.arun(self._io_save_json_unlocked, filepath) - def _io_save_json_unlocked(self, filepath: str | Path, include_audit_log: bool = True) -> str: - save_investigation_json(self._main_investigation, filepath, include_audit_log=include_audit_log) + def _io_save_json_unlocked(self, filepath: str | Path) -> str: + save_investigation_json(self._main_investigation, filepath) return str(Path(filepath).resolve()) diff --git a/tests/test_cli.py b/tests/test_cli.py index 1dc6e8f..82ae512 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -201,31 +201,6 @@ def test_display_summary_exclude_levels() -> None: assert "none_finding" in output_all -def test_display_summary_audit_log_table() -> None: - """display_summary shows audit log when requested.""" - from decimal import Decimal - from io import StringIO - - from rich.console import Console - - from cyvest.io_rich import display_summary - - cv = Cyvest() - obs = cv.observable(Cyvest.OBS.URL, "https://example.com", internal=False) - obs.with_ti("virustotal", score=Decimal("6.0"), level=Cyvest.LVL.MALICIOUS) - cv.finding("score-finding", "test", "Score finding").with_score(Decimal("1.0"), reason="initial").with_score( - Decimal("2.0"), reason="bump" - ) - - output = StringIO() - console = Console(file=output, width=140) - display_summary(cv, console.print, show_graph=False, show_audit_log=True) - rendered = output.getvalue() - - assert "Audit Log" in rendered - assert "virustotal" in rendered - - def test_cli_diff_no_differences(tmp_path: Path) -> None: """CLI 'diff' command succeeds for identical investigations.""" from decimal import Decimal diff --git a/tests/test_cyvest.py b/tests/test_cyvest.py index 777ad72..06f937c 100644 --- a/tests/test_cyvest.py +++ b/tests/test_cyvest.py @@ -705,74 +705,6 @@ def test_io_to_invest_serialization() -> None: assert data["observables"][obs.key]["value"] == "https://malicious.com" -def test_io_to_invest_include_audit_log_default() -> None: - """Test that io_to_invest includes audit_log by default.""" - cv = Cyvest() - cv.observable_create(Cyvest.OBS.URL, "https://example.com") - - schema = cv.io_to_invest() - - # By default, audit_log should be a list with events - assert schema.audit_log is not None - assert isinstance(schema.audit_log, list) - assert len(schema.audit_log) > 0 # At least INVESTIGATION_STARTED event - - -def test_io_to_invest_exclude_audit_log() -> None: - """Test that io_to_invest can exclude audit_log.""" - cv = Cyvest() - cv.observable_create(Cyvest.OBS.URL, "https://example.com") - - schema = cv.io_to_invest(include_audit_log=False) - - # audit_log should be None when disabled - assert schema.audit_log is None - - # Verify JSON output has null - data = schema.model_dump(mode="json", by_alias=True) - assert data["audit_log"] is None - - -def test_io_save_json_exclude_audit_log(tmp_path) -> None: - """Test that io_save_json can exclude audit_log from output.""" - import json - - cv = Cyvest() - cv.observable_create(Cyvest.OBS.DOMAIN, "test.com") - - filepath = tmp_path / "investigation.json" - - # Save without audit_log - cv.io_save_json(str(filepath), include_audit_log=False) - - # Verify JSON content - with open(filepath) as f: - data = json.load(f) - - assert data["audit_log"] is None - - -def test_io_save_json_include_audit_log_default(tmp_path) -> None: - """Test that io_save_json includes audit_log by default.""" - import json - - cv = Cyvest() - cv.observable_create(Cyvest.OBS.DOMAIN, "test.com") - - filepath = tmp_path / "investigation.json" - - # Save with default (include audit_log) - cv.io_save_json(str(filepath)) - - # Verify JSON content - with open(filepath) as f: - data = json.load(f) - - assert data["audit_log"] is not None - assert isinstance(data["audit_log"], list) - assert len(data["audit_log"]) > 0 - - def test_io_to_markdown_generates_report() -> None: """Test Markdown report generation.""" cv = Cyvest() diff --git a/tests/test_score_algorithm.py b/tests/test_score_algorithm.py index 9e7283e..22ea99a 100644 --- a/tests/test_score_algorithm.py +++ b/tests/test_score_algorithm.py @@ -1,8 +1,7 @@ """ Tests for the reworked score algorithm. -Tests MAX vs SUM modes, finding score calculation, hierarchical propagation, -and audit log access. +Tests MAX vs SUM modes, finding score calculation, and hierarchical propagation. """ from decimal import Decimal @@ -224,81 +223,6 @@ def test_finding_score_preserves_higher_current_score() -> None: assert finding.level == Cyvest.LVL.MALICIOUS -def test_observable_score_audit_events() -> None: - """Observable score changes should appear in the audit log.""" - cv = Cyvest() - - # Create observable - obs = cv.observable_create(Cyvest.OBS.IPV4, "10.0.0.1") - - # Initial audit log should have no score events for this observable. - events = [ - event - for event in cv.investigation_get_audit_log() - if event.object_key == obs.key and event.event_type.startswith("SCORE") - ] - assert len(events) == 0 - - # Add threat intel (triggers score change) - cv.observable_add_threat_intel(obs.key, source="source1", score=Decimal("5.0")) - - # Audit log should now have a score change entry. - events = [ - event - for event in cv.investigation_get_audit_log() - if event.object_key == obs.key and event.event_type.startswith("SCORE") - ] - assert len(events) == 1 - assert events[0].details["old_score"] == 0.0 - assert events[0].details["new_score"] == 5.0 - assert events[0].details["old_level"] == Cyvest.LVL.INFO.value - assert events[0].details["new_level"] == Cyvest.LVL.MALICIOUS.value - assert "source1" in (events[0].reason or "") - - # Add another threat intel - cv.observable_add_threat_intel(obs.key, source="source2", score=Decimal("8.0")) - - # Audit log should now have 2 entries - events = [ - event - for event in cv.investigation_get_audit_log() - if event.object_key == obs.key and event.event_type.startswith("SCORE") - ] - assert len(events) == 2 - assert events[1].details["old_score"] == 5.0 - assert events[1].details["new_score"] == 8.0 - assert "source2" in (events[1].reason or "") - - -def test_finding_score_audit_events() -> None: - """Finding score changes should appear in the audit log.""" - cv = Cyvest() - - # Create finding - finding = cv.finding_create("finding1", "test", "Test finding") - - events = [ - event - for event in cv.investigation_get_audit_log() - if event.object_key == finding.key and event.event_type.startswith("SCORE") - ] - assert len(events) == 0 - - # Create observable and link to finding - obs = cv.observable_create(Cyvest.OBS.IPV4, "10.0.0.1") - cv.observable_add_threat_intel(obs.key, source="source1", score=Decimal("5.0")) - cv.finding_link_observable(finding.key, obs.key) - - # Audit log should now have entries from score updates - events = [ - event - for event in cv.investigation_get_audit_log() - if event.object_key == finding.key and event.event_type.startswith("SCORE") - ] - assert len(events) >= 1 - assert any(entry.details.get("new_score") == 5.0 for entry in events) - - def test_score_propagation_through_hierarchy() -> None: """Test score propagation through multi-level hierarchy in MAX mode.""" cv = Cyvest(score_mode_obs=ScoreMode.MAX) @@ -598,14 +522,6 @@ def test_safe_level_score_updates_with_frozen_level() -> None: assert obs.score == Decimal("0") assert obs.level == Cyvest.LVL.SAFE # Level frozen at SAFE even though score=0 would be INFO - # Verify audit log tracked the score change - events = [ - event - for event in cv.investigation_get_audit_log() - if event.object_key == obs.key and event.event_type.startswith("SCORE") - ] - assert len(events) > 0 - def test_non_safe_levels_recalculate_and_can_downgrade() -> None: """Non-SAFE levels are recalculated from score and can downgrade.""" diff --git a/tests/test_v6_models.py b/tests/test_v6_models.py index 5a86d9c..157e172 100644 --- a/tests/test_v6_models.py +++ b/tests/test_v6_models.py @@ -142,7 +142,6 @@ def test_migrate_v5_to_v6_rewrites_findings_and_preserves_email() -> None: "level": "NONE", "whitelisted": False, "whitelists": [], - "audit_log": [], "observables": { "obs:file:root": { "type": "file",