Schemas

Schemas are the typed message objects used to transmit data between components (Parser → Detector). Each schema class wraps a Protobuf message and provides a small, convenient Python API for creation, inspection, (de)serialization and validation.

This document summarizes the available schema classes, the BaseSchema API and common usage patterns.

Design goals

• Strongly-typed contracts between components.
• Lightweight wrapper over generated Protobuf types.
• Simple API for tests and runtime wiring.
• Safe (de)serialization for transport and persistence.

Architecture

Schemas are used to transfer data between components. Each schema implements the methods defined in BaseSchema. This class acts as a wrapper around the underlying Protobuf classes (op.SchemaT).

All concrete schema classes inherit from BaseSchema. Key utility methods:

class BaseSchema:
    def __contains__(self, idx: str) -> bool:
        """Return if a variable is in the schema"""

    def as_dict(self) -> dict[str, Any]:
        """Return the schema variables as a dictionary."""

    def get_schema(self) -> op.SchemaT:
        """Retrieve the current schema instance."""

    def set_schema(self, schema: op.SchemaT) -> None:
        """Set the schema instance and update attributes."""

    def init_schema(self, kwargs: dict[str, Any] | None) -> None:
        """Initialize the schema instance and set attributes."""

    def is_field_list(self, field_name: str) -> bool:
        """Check if a field is a list."""

    def copy(self) -> "BaseSchema":
        """Create a deep copy of the schema instance."""

    def serialize(self) -> bytes:
        """Serialize the schema instance to bytes."""

    def deserialize(self, message: bytes) -> None:
        """Deserialize bytes to populate the schema instance.

        Raises NotSupportedSchema if the bytes cannot be parsed.
        """

    def check_is_same(self, other: Self) -> None:
        """Check if another schema instance is of the same schema type.

        Raises IncorrectSchema if it is not.
        """

    def __eq__(self, other: object) -> bool:
        """Check equality between two schema instances."""

The exceptions these methods raise can be imported from detectmatelibrary.schemas:

from detectmatelibrary.schemas import IncorrectSchema, NotSupportedSchema

Schema Classes

Below are the primary schema classes and their main fields. All fields are optional at the Protobuf level; components should document which fields they require.

LogSchema

Represents a raw log message.

Fields:

Field	Type	Notes
logID	string	Unique identifier for the raw log.
log	string	Raw log text.
logSource	string	Source of the log (file, topic, etc.).
hostname	string	Hostname where log originated.

ParserSchema

Output of a Parser. Contains parsed fields and template information.

Fields:

Field	Type	Notes
parserType	string	Parser type.
parserID	string	Parser instance identifier.
eventID	int32	Template/event identifier.
template	string	Event template text.
variables	repeated string	Parameters extracted from the template.
parsedLogID	string	ID assigned after parsing (optional).
logID	string	Original raw log ID (link to LogSchema).
log	string	Raw log text.
headerVariables	map	Key/value pairs from format extraction.
receivedTimestamp	int64	Timestamp when log was received.
parsedTimestamp	int64	Timestamp when parsing completed.

DetectorSchema

Output from Detectors (alerts / findings).

Fields:

Field	Type	Notes
detectorID	string	Detector instance identifier.
detectorType	string	Type/name of detector.
alertID	string	Unique alert identifier.
detectedTimestamp	int64	When the alert was produced.
logIDs	repeated string	IDs of logs related to the alert.
score	float	Confidence/score (if applicable).
logTimestamps	repeated int64	Timestamps extracted from logs.
description	string	Human-readable description of the alert.
receivedTimestamp	int64	When inputs were received by detector.
alertDetails	map	Additional alert metadata.

AggregateSchema

Output from Alert Aggregations (alerts / findings).

Fields:

Field	Type	Notes
detectorIDs	repeated string	List of detector instance identifier.
detectorTypes	repeated string	List of type/name of detectors.
alertIDs	repeated string	List of unique alert identifiers.
aggregatedTimestamp	int64	When the aggregation was produced.
logIDs	repeated string	IDs of logs related to the alerts.
logTimestamps	repeated int64	Timestamps extracted from logs.
description	string	Human-readable description of the alert aggregation.
alertDetails	map	Additional alert metadata from the alert aggregation.

Examples

Short examples of working with the schemas.

Initialize a schema

from detectmatelibrary import schemas


log_dict = {"log": "Test log"}

log_schema = schemas.LogSchema(log_dict)
print(log_schema.log == "Test log")  # True

Assign values

log_schema = schemas.LogSchema()
log_schema.log = "Test log"
print(log_schema["log"] == log_schema.log)  # True

log_schema2 = schemas.LogSchema()
print(log_schema == log_schema2)  # False

log_schema2.log = "Test log"
print(log_schema == log_schema2)  # True

Serialization

log_schema = schemas.LogSchema()
log_schema.log = "Test log"
serialized = log_schema.serialize()
print(isinstance(serialized, bytes))  # True

new_log_schema = schemas.LogSchema()
new_log_schema.deserialize(serialized)
print(new_log_schema == log_schema)  # True