Metadata-Version: 2.5
Name: ocsf-emitter
Version: 0.1.3
Summary: Construct, validate, and emit OCSF 1.1.0 events (Detection Finding, Compliance Finding, Authentication, Account Change, OS Patch State, API Activity, Web Resources Activity, File Hosting Activity) with a consistent shape.
Project-URL: Homepage, https://github.com/Hacker0x01/ocsf-emitter
Project-URL: Documentation, https://hacker0x01.github.io/ocsf-emitter/
Project-URL: Repository, https://github.com/Hacker0x01/ocsf-emitter
Project-URL: Issues, https://github.com/Hacker0x01/ocsf-emitter/issues
Author: HackerOne
License-Expression: MIT
License-File: LICENSE
Keywords: detection,finding,ocsf,security-lake,siem
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic<3,>=2.6
Provides-Extra: codegen
Requires-Dist: datamodel-code-generator[http]>=0.25; extra == 'codegen'
Requires-Dist: ocsf-lib<1,>=0.10; extra == 'codegen'
Provides-Extra: securitylake
Requires-Dist: pyarrow>=14; extra == 'securitylake'
Description-Content-Type: text/markdown

# ocsf-emitter

Construct, validate, and emit **OCSF 1.1.0** events with a consistent shape and
mandatory runtime validation. Supports eight classes: Detection Finding (2004),
Compliance Finding (2003), Authentication (3002), Account Change (3001),
Operating System Patch State (5004), API Activity (6003), Web Resources Activity
(6001), and File Hosting Activity (6006).

This is an internal library. Other services import it to turn their own
detection signals into valid OCSF events; the library owns the OCSF field
names, the schema-version pin, the house defaults, and validation. **Transport
is deliberately out of scope** -- `emit()` returns a validated,
JSON-serializable payload and the caller ships it however it likes.

Each class has a builder (`build_detection_finding`, `build_authentication`,
`build_file_hosting`, …) sharing a common keyword shape; `emit()`/`validate()`
accept any supported class. See the [usage guide](docs/usage.md) for the full
class table and examples.

## Install

```bash
uv pip install -e .                  # runtime: pydantic only
uv pip install -e ".[securitylake]"  # + pyarrow, for the Parquet writer
uv pip install -e ".[dev]"           # + mypy, pytest, ruff (and pyarrow)
uv pip install -e ".[codegen]"       # + tools to regenerate the OCSF models
```

## Usage

```python
import ocsf_emitter
from ocsf_emitter import (
    build_detection_finding, emit,
    Severity, Status, Activity, Confidence, RiskLevel,
    Observable, ObservableType, MitreAttack,
)

# Configure the emitting product once at startup (see "Product identity").
ocsf_emitter.configure_product(name="Example Detector", vendor_name="Example, Inc.")

finding = build_detection_finding(
    uid="det-2026-0715-001",             # stable id -> finding_info.uid
    title="Impossible-travel login",
    severity=Severity.HIGH,              # our enum -> OCSF severity_id
    message="User alice logged in from two continents within 4 minutes.",
    status=Status.NEW,                   # -> status_id
    activity=Activity.CREATE,            # -> activity_id + type_uid
    observables=[
        Observable(ObservableType.USER_NAME, "alice"),
        Observable(ObservableType.IP_ADDRESS, "203.0.113.7"),
    ],
    confidence=Confidence.HIGH,          # -> confidence_id
    risk_level=RiskLevel.HIGH,           # -> risk_level_id
    attacks=[MitreAttack("T1078", "Valid Accounts", "TA0001", "Initial Access")],
)

payload = emit(finding)   # validates, returns dict; raises InvalidFindingError if invalid
```

`build_detection_finding(...)` returns a **typed, already-valid**
`DetectionFinding` model instance. `emit(...)` runs full validation and returns
a `dict`; `emit_json(...)` returns a JSON string. `build_from_signal(signal)`
takes a `DetectionSignal` dataclass if you'd rather build the domain object
yourself.

See [`tests/golden_detection_finding.json`](tests/golden_detection_finding.json)
for a full sample payload you can eyeball against the OCSF `detection_finding`
spec.

## How validation behaves

Validation is **mandatory and automatic** inside `emit()`/`emit_json()`, and is
also callable standalone via `validate(finding)`. It does two things:

1. **Schema validation** -- re-runs Pydantic validation over the finding's
   current field values (catching any mutation after construction).
2. **OCSF invariant checks** -- looks the event's class up in the class registry
   by its `class_uid` and verifies `category_uid`, the `class_name`/
   `category_name` siblings, `type_uid == class_uid*100 + activity_id`, and that
   `metadata.version` matches the pinned schema version.

On failure it raises `InvalidFindingError`, whose message and `.field_errors`
list **name the offending field(s)**:

```
ocsf_emitter.errors.InvalidFindingError: Detection finding failed OCSF schema validation
  - severity_id: Input should be 0, 1, 2, 3, 4, 5, 6 or 99
```

## Model layer: chosen path and rationale

We generate a **full Pydantic v2 model tree** from OCSF's JSON Schema using
[`datamodel-code-generator`](https://github.com/koxudaxi/datamodel-code-generator),
and **commit the result** to
[`src/ocsf_emitter/_models.py`](src/ocsf_emitter/_models.py). At runtime the
package depends only on `pydantic` -- no network access, no code generation.

Why this rather than `py-ocsf-models`? The task allowed either; we chose
schema-generation for two reasons:

- **Exact fidelity to a pinned OCSF version.** The models come straight from
  the OCSF schema for the exact version we pin -- no dependency on a third
  party's release cadence for class coverage.
- **Self-contained and auditable.** The generated module is committed and
  reviewable, and regeneration is a single script.

### How the pinned version is generated

We must pin an **older** OCSF version (1.1.0 -- see Security Lake below), and
the OCSF JSON Schema HTTP endpoint only serves the *latest* deployed version. So
`scripts/gen_models.py`:

1. Fetches the pinned version's **metaschema** with `ocsf-lib`
   (`OcsfApiClient().get_schema("1.1.0")`) -- this works for any version.
2. Converts that metaschema (every class in `ROOT_CLASSES` plus the **union** of
   their transitive object closures) into a self-contained draft **JSON Schema**,
   where each root class and shared object is a `$def` so codegen emits one
   Pydantic model each. Attributes tagged with an OCSF `profile` (e.g. `cloud`,
   `osint`) are dropped so we get the **base** classes -- mirroring the schema
   server's `?profiles=` selector, and keeping profile-only fields out of the
   required list.
3. Feeds that JSON Schema to `datamodel-code-generator` -> Pydantic v2.

## Bumping the OCSF schema version

> **Before bumping:** if you ship to AWS Security Lake, confirm the target
> version is one Security Lake accepts (see below). The CI `aws-validation` job
> is a hard gate and will fail for unsupported versions.

1. Regenerate the models for the desired version:

   ```bash
   uv run --extra codegen python scripts/gen_models.py 1.1.0
   ```

2. Update the **one-line** pin in
   [`src/ocsf_emitter/defaults.py`](src/ocsf_emitter/defaults.py) to match:

   ```python
   OCSF_SCHEMA_VERSION = "1.1.0"   # <- change this
   ```

3. Run the suite and review the golden diff:

   ```bash
   uv run pytest && uv run mypy && uv run ruff check .
   ```

   If the emitted shape changed intentionally, regenerate
   `tests/golden_detection_finding.json` and review the diff. Note that a newer
   version may reintroduce `RootModel`/union wrappers on some objects, which
   would require builder adjustments.

## AWS Security Lake compatibility

This package targets **AWS Security Lake custom-source** ingestion.

**Version.** Security Lake custom sources accept OCSF **1.1.0 / 1.0.0-rc.2**, and
the [AWS OCSF validation tool](https://github.com/aws-samples/amazon-security-lake-ocsf-validation)
maps `detection_finding` only under **1.1.0**. We therefore pin **1.1.0**
(not a newer version). To keep findings acceptable, the builder also stamps the
sibling label fields Security Lake expects: `class_name` (`"Detection Finding"`)
and `category_name` (`"Findings"`).

**Validation is proven against AWS's own tool.** CI runs that tool against an
emitted finding as a **blocking** job
([`.github/workflows/ci.yml`](.github/workflows/ci.yml)), pinned at a known
commit. Locally:

```bash
git clone https://github.com/aws-samples/amazon-security-lake-ocsf-validation.git /tmp/aws-ocsf
uv pip install -e ".[securitylake]" && uv pip install -r /tmp/aws-ocsf/requirements.txt pytest
OCSF_AWS_VALIDATION_DIR=/tmp/aws-ocsf uv run pytest tests/test_integ_aws_validation.py -v
```

**Parquet packaging.** Security Lake wants **Parquet** objects (not JSON),
zstd-compressed, partitioned in S3 and sorted by time. The optional
`ocsf_emitter.securitylake` module (install the `securitylake` extra) does this
without touching S3 -- you upload the returned bytes at the returned key:

```python
from ocsf_emitter import securitylake as sl

obj = sl.build_parquet_object(
    findings,                       # a batch of built findings
    source_location="my_detector",  # the prefix Security Lake assigned you
    region="us-east-1",
    account_id="123456789012",
    object_name="batch-001",
)
# obj.key  -> ext/my_detector/region=us-east-1/accountId=.../eventDay=YYYYMMDD/batch-001.parquet
# obj.data -> Parquet bytes (zstd); upload to your Security Lake S3 bucket at obj.key
```

The writer validates every finding (via `emit`), sorts records by `time`, uses
zstd compression, and bounds data-page (<=1 MB) and row-group sizes per the
Security Lake custom-source requirements.

## Product identity

The emitting product is **configurable**, not hardcoded, so any service can use
this library. Set it once at startup:

```python
ocsf_emitter.configure_product(name="My Service", vendor_name="My Org")
```

or pass `product=ocsf_emitter.make_product(...)` per call. Building a finding
with no product configured raises `OcsfEmitterError` rather than emitting an
unattributed finding.

## Package layout

```
src/ocsf_emitter/
  __init__.py     public API: build_detection_finding, build_from_signal, emit, ...
  domain.py       our domain shapes: DetectionSignal, Observable, MitreAttack, enums
  builders.py     domain signal  ->  OCSF DetectionFinding
  defaults.py     schema-version pin, metadata/product, severity/status/... mappings
  validate.py     runtime validation; raises InvalidFindingError
  emit.py         serialize to JSON dict/str (transport-agnostic)
  securitylake.py OCSF -> Parquet for AWS Security Lake (needs [securitylake] extra)
  errors.py       OcsfEmitterError, InvalidFindingError
  _models.py      GENERATED OCSF Pydantic models (do not edit by hand)
scripts/gen_models.py   regenerate _models.py (metaschema -> JSON Schema -> Pydantic)
tests/                  mappings, builder/emit, validation-rejection, golden, parquet
tests/test_integ_aws_validation.py   runs the AWS OCSF validation tool (CI-gated)
.github/workflows/ci.yml              unit job + blocking AWS-validation job
```

## Development

```bash
uv run pytest        # tests
uv run mypy          # strict type-checking
uv run ruff check .  # lint
uv run ruff format . # format
```
