Security Lake

securitylake

AWS Security Lake packaging for OCSF Detection Findings.

Security Lake custom sources must deliver Parquet objects (not JSON), partitioned in S3 by region / account / event-day, with records sorted by time. This module turns validated findings into a Parquet byte payload and computes the S3 object key. It does NOT talk to S3 -- the caller uploads the bytes at the returned key, keeping transport out of the package (see README).

Requirements implemented (per the Security Lake custom-source docs): * Apache Parquet, zstandard compression. * Data page size <= 1 MB (uncompressed); row group size <= 256 MB. * Records sorted by time within each object. * Partition prefix: ext/{source}/region={r}/accountId={a}/eventDay={YYYYMMDD}/.

This module requires the securitylake extra (pyarrow): uv pip install -e ".[securitylake]"

ParquetObject dataclass

ParquetObject(key: str, data: bytes, record_count: int)

A ready-to-upload Security Lake object: Parquet data at S3 key.

partition_prefix

partition_prefix(
    *,
    source_location: str,
    region: str,
    account_id: str,
    event_day: str,
) -> str

Build the Security Lake S3 partition prefix.

ext/{source_location}/region={region}/accountId={account_id}/eventDay={event_day}/

Source code in src/ocsf_emitter/securitylake.py

def partition_prefix(*, source_location: str, region: str, account_id: str, event_day: str) -> str:
    """Build the Security Lake S3 partition prefix.

    ``ext/{source_location}/region={region}/accountId={account_id}/eventDay={event_day}/``
    """
    return f"ext/{source_location}/region={region}/accountId={account_id}/eventDay={event_day}/"

event_day_from_ms

event_day_from_ms(time_ms: int) -> str

Convert an epoch-ms timestamp to a Security Lake eventDay (UTC YYYYMMDD).

Source code in src/ocsf_emitter/securitylake.py

def event_day_from_ms(time_ms: int) -> str:
    """Convert an epoch-ms timestamp to a Security Lake ``eventDay`` (UTC YYYYMMDD)."""
    return datetime.fromtimestamp(time_ms / 1000, tz=UTC).strftime("%Y%m%d")

to_parquet_bytes

to_parquet_bytes(
    findings: Sequence[DetectionFinding],
) -> bytes

Validate, sort by time, and serialize findings to a Parquet byte string.

Source code in src/ocsf_emitter/securitylake.py

def to_parquet_bytes(findings: Sequence[_m.DetectionFinding]) -> bytes:
    """Validate, sort by time, and serialize findings to a Parquet byte string."""
    if not findings:
        raise ValueError("cannot build a Parquet object from zero findings")
    payloads = _validated_payloads(findings)
    table = pa.Table.from_pylist(payloads)

    import io

    buf = io.BytesIO()
    pq.write_table(
        table,
        buf,
        compression=_COMPRESSION,
        data_page_size=_DATA_PAGE_SIZE_BYTES,
        row_group_size=_ROW_GROUP_SIZE_ROWS,
    )
    return buf.getvalue()

build_parquet_object

build_parquet_object(
    findings: Sequence[DetectionFinding],
    *,
    source_location: str,
    region: str,
    account_id: str,
    object_name: str,
    event_day: str | None = None,
) -> ParquetObject

Package findings into a single Security Lake Parquet object.

Parameters:

Name	Type	Description	Default
findings	Sequence[DetectionFinding]	One or more validated-on-emit detection findings. All should share an event day; event_day defaults to the day of the earliest finding's time.	required
source_location	str	The unique prefix Security Lake assigned to this source.	required
region	str	AWS region the data is uploaded to (e.g. us-east-1).	required
account_id	str	AWS account id the records pertain to (or external...).	required
object_name	str	File name for the object (.parquet appended if absent).	required
event_day	str | None	YYYYMMDD UTC partition day. Defaults to the earliest finding's event day.	None

Returns:

Name	Type	Description
A	ParquetObject	class:ParquetObject with the S3 key and Parquet data bytes.

Source code in src/ocsf_emitter/securitylake.py

def build_parquet_object(
    findings: Sequence[_m.DetectionFinding],
    *,
    source_location: str,
    region: str,
    account_id: str,
    object_name: str,
    event_day: str | None = None,
) -> ParquetObject:
    """Package findings into a single Security Lake Parquet object.

    Args:
        findings: One or more validated-on-emit detection findings. All should
            share an event day; ``event_day`` defaults to the day of the
            earliest finding's ``time``.
        source_location: The unique prefix Security Lake assigned to this source.
        region: AWS region the data is uploaded to (e.g. ``us-east-1``).
        account_id: AWS account id the records pertain to (or ``external...``).
        object_name: File name for the object (``.parquet`` appended if absent).
        event_day: ``YYYYMMDD`` UTC partition day. Defaults to the earliest
            finding's event day.

    Returns:
        A :class:`ParquetObject` with the S3 ``key`` and Parquet ``data`` bytes.
    """
    if not findings:
        raise ValueError("cannot build a Parquet object from zero findings")

    data = to_parquet_bytes(findings)

    if event_day is None:
        earliest = min(int(f.time) for f in findings)
        event_day = event_day_from_ms(earliest)

    if not object_name.endswith(".parquet"):
        object_name = f"{object_name}.parquet"

    prefix = partition_prefix(
        source_location=source_location,
        region=region,
        account_id=account_id,
        event_day=event_day,
    )
    return ParquetObject(key=prefix + object_name, data=data, record_count=len(findings))