OCSF Schema Validation
# TODO narrow down the *

from .errors import *
from .processor import process_includes
from .reader import DictReader, FileReader, Reader
from .runner import ValidationRunner, ValidatorOptions
from .types import *
from .validators import *

ocsf_validator/__init__.py
query-ai/ocsf-validator
from argparse import ArgumentParser

from ocsf_validator.reader import FileReader, ReaderOptions
from ocsf_validator.runner import ValidationRunner, ValidatorOptions

parser = ArgumentParser(prog="ocsf-validator", description="OCSF Schema Validation")
parser.add_argument("path", help="The OCSF schema root directory", action="store")
args = parser.parse_args()

opts = ValidatorOptions(base_path=args.path)

validator = ValidationRunner(opts)

validator.validate()

ocsf_validator/__main__.py
query-ai/ocsf-validator
from typing import Iterable, Optional, TypeVar

TCollector = TypeVar("TCollector", bound="Collector")


class Collector:
    """An error collector used by schema operations so that encountered errors
    can be saved for later and displayed as test output.

    The default behavior is to `raise` all exceptions, so you shouldn't
    notice the collector until `throw` is `False`.
    """

    default: TCollector  # type: ignore
    """Simple singleton used whenever an Optional[Collector] parameter is None."""

    def __init__(self, throw: bool = True):
        self._exceptions: list[Exception] = []
        self._throw = throw

    def handle(self, err: Exception):
        """Handle an exception.

        By default, exceptions are stored and raised. But if `throw` is `False`,
        exceptions will only be stored for later."""

        self._exceptions.append(err)
        if self._throw:
            raise err

    def exceptions(self):
        return self._exceptions

    def flush(self):
        e = list(self._exceptions)
        self._exceptions = []
        return e

    def __len__(self):
        return len(self._exceptions)

    def __iter__(self) -> Iterable[Exception]:
        return iter(self._exceptions)


Collector.default = Collector()


class ValidationError(Exception):
    """Base class for validation errors."""

    ...


class InvalidBasePathError(ValidationError):
    ...


class InvalidMetaSchemaError(ValidationError):
    ...


class UnusedAttributeError(ValidationError):
    def __init__(self, attr: str):
        self.attr = attr
        super().__init__(f"Unused attribute {attr}")


class MissingRequiredKeyError(ValidationError):
    def __init__(
        self,
        key: str,
        file: str,
        cls: Optional[type] = None,
        trail: Optional[list[str]] = None,
    ):
        self.key = key
        self.file = file
        self.cls = cls
        self.trail = trail

        if trail is None:
            trail_str = ""
        else:
            trail_str = ".".join(trail)

        if cls is None:
            cls_str = ""
        else:
            cls_str = cls.__name__ + "."

        super().__init__(
            f"Missing required key `{cls_str}{key}` at `{trail_str}` in {file}"
        )


class UnknownKeyError(ValidationError):
    def __init__(
        self,
        key: str,
        file: str,
        cls: Optional[type] = None,
        trail: Optional[list[str]] = None,
    ):
        self.key = key
        self.file = file
        self.cls = cls
        self.trail = trail

        if trail is None:
            trail_str = ""
        else:
            trail_str = ".".join(trail)

        if cls is None:
            cls_str = ""
        else:
            cls_str = cls.__name__

        super().__init__(
            f"Unrecognized key `{key}` of `{cls_str}` at `{trail_str}` in {file}"
        )


class DependencyError(ValidationError):
    def __init__(self, file: str, include: str, message: Optional[str] = None):
        self.file = file
        self.include = include
        super().__init__(message)


class MissingIncludeError(DependencyError):
    def __init__(self, file: str, include: str):
        self.file = file
        self.include = include
        super().__init__(file, include, f"Missing include target '{include}' in {file}")


class MissingBaseError(DependencyError):
    def __init__(self, file: str, include: str):
        self.file = file
        self.include = include
        super().__init__(file, include, f"Missing base record '{include}' in {file}")


class ImpreciseBaseError(DependencyError):
    def __init__(self, file: str, include: str):
        self.file = file
        self.include = include
        super().__init__(
            file,
            include,
            f"Possibly ambiguous base record definition '{include}' in {file}",
        )


class MissingProfileError(DependencyError):
    def __init__(self, file: str, include: str):
        self.file = file
        self.include = include
        super().__init__(file, include, f"Missing profile '{include}' in {file}")


class SelfInheritanceError(DependencyError):
    def __init__(self, file: str, include: str):
        self.file = file
        self.include = include
        super().__init__(file, include, f"Inheritance from self '{include}' in {file}")


class RedundantProfileIncludeError(DependencyError):
    def __init__(self, file: str, include: str):
        self.file = file
        self.include = include
        super().__init__(
            file,
            include,
            f"Redundant $include and profiles entry '{include}' in {file}",
        )


class UndetectableTypeError(ValidationError):
    def __init__(self, file: str):
        self.file = file
        super().__init__(f"Unable to detect type of {file}")


class IncludeTypeMismatchError(ValidationError):
    def __init__(
        self, file: str, include: str, t: type | str, directive: str = "$include"
    ):
        self.file = file
        self.include = include
        if isinstance(t, str):
            self.cls = t
        else:
            self.cls: str = t.__name__
        self.directive = directive
        super().__init__(
            f"`{directive}` type mismatch in {file}: expected type `{self.cls}` for {include}"
        )


class TypeNameCollisionError(ValidationError):
    def __init__(self, name: str, kind: str, file1: str, file2: str):
        self.name = name
        self.kind = kind
        self.file1 = file1
        self.file2 = file2
        super().__init__(
            f"Name collision for `{name}` within `{kind}` between {file1} and {file2}"
        )


class UndefinedAttributeError(ValidationError):
    def __init__(self, attr: str, file: str):
        self.attr = attr
        self.file = file
        super().__init__(
            f"Attribute `{attr}` in {file} is not defined in dictionary.json"
        )

ocsf_validator/errors.py
query-ai/ocsf-validator
import re
from abc import ABC
from pathlib import Path
from typing import Optional

from ocsf_validator.types import *


class Matcher:
    def match(self, value: str) -> bool:
        raise NotImplementedError()

    @staticmethod
    def make(pattern):
        if isinstance(pattern, Matcher):
            return pattern
        else:
            return RegexMatcher(pattern)


class TypeMatcher:
    def get_type(self) -> type:
        raise NotImplementedError()


class AnyMatcher(Matcher):
    def __init__(self, matchers: Optional[list[Matcher]] = None):
        if matchers is not None:
            self._matchers = matchers
        else:
            self._matchers = []

    def match(self, value: str):
        for matcher in self._matchers:
            if matcher.match(value):
                return True

    def add(self, matcher: Matcher):
        self._matchers.append(matcher)


class RegexMatcher(Matcher):
    def __init__(self, pattern: str | re.Pattern):
        if isinstance(pattern, str):
            self._pattern = re.compile(pattern)
        else:
            self._pattern = pattern

    def match(self, value: str):
        return self._pattern.match(value) is not None


class GlobMatcher(Matcher):
    def __init__(self, pattern: str):
        self._pattern = pattern

    def match(self, value: str):
        path = Path(value)
        return path.match(self._pattern)


class DictionaryMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*dictionary.json")

    def get_type(self):
        return OcsfDictionary


class VersionMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*version.json")

    def get_type(self):
        return OcsfVersion


class ObjectMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*objects/.*json")

    def get_type(self):
        return OcsfObject


class EventMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*events/.*json")

    def get_type(self):
        return OcsfEvent


class ExtensionMatcher(GlobMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = "extensions/*/extension.json"

    def get_type(self):
        return OcsfExtension


class IncludeMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*includes/.*.json")

    def get_type(self):
        return OcsfInclude


class ProfileMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*profiles/.*.json")

    def get_type(self):
        return OcsfProfile


class CategoriesMatcher(RegexMatcher, TypeMatcher):
    def __init__(self):
        self._pattern = re.compile(r".*categories.json")

    def get_type(self):
        return OcsfCategories

ocsf_validator/matchers.py
query-ai/ocsf-validator
from pathlib import Path
from typing import Any, Callable, Optional

from ocsf_validator.errors import *
from ocsf_validator.reader import Reader
from ocsf_validator.type_mapping import TypeMapping
from ocsf_validator.types import (ATTRIBUTES_KEY, EXTENDS_KEY, INCLUDE_KEY,
                                  PROFILES_KEY, OcsfDictionary, OcsfEvent,
                                  OcsfObject)


def deep_merge(
    subj: dict[str, Any], other: dict[str, Any], exclude: Optional[set[str]] = None
):
    """Recursive merging of dictionary keys.

    `subj | other` is more readable, but it doesn't merge recursively. If
    subj and other each have an "attributes" key with a dictionary value,
    only the first "attributes" dictionary will be present in the resulting
    dictionary. And thus this recursive merge."""

    if exclude is None:
        skip = set()
    else:
        skip = exclude

    for k, v in other.items():
        if k not in skip:
            if k in subj and isinstance(v, dict):
                deep_merge(subj[k], other[k])

            elif k not in subj:
                subj[k] = other[k]


def exclude_props(t1: type, t2: type):
    if not hasattr(t1, "__annotations__") or not hasattr(t2, "__annotations__"):
        raise Exception("Unexpected types in comparison")
    s1 = set(t1.__annotations__.keys())
    s2 = set(t2.__annotations__.keys())
    return s2 - s1


class DependencyResolver:
    def __init__(self, reader: Reader, types: TypeMapping):
        self._reader = reader
        self._types = types

    def resolve_include(
        self, target: str, relative_to: Optional[str] = None
    ) -> str | None:
        """Find a file from an OCSF $include directive.

        For a given file f, search:
          extn/f
          extn/f.json
          f
          f.json
        """

        filenames = [target]
        if Path(target).suffix != ".json":
            filenames.append(target + ".json")

        for file in filenames:
            if relative_to is not None:
                # Search extension for relative include path, e.g. /includes/thing.json -> /extensions/stuff/includes/thing.json
                extn = self._types.extension(relative_to)
                if extn is not None:
                    k = self._reader.key("extensions", extn, file)
                    if k in self._reader:
                        return k

            k = self._reader.key(file)
            if k in self._reader:
                return k

        return None

    def resolve_profile(self, profile: str, relative_to: str) -> str | None:
        """Find a file from an OCSF profiles directive.

        For a requested profile p, search:
          extn/profiles/p
          extn/profiles/p.json
          profiles/p
          profiles/p.json
          extn/p
          extn/p.json
          p
          p.json
        """
        file = self.resolve_include(profile, relative_to)
        if file is None:
            path = str(Path("profiles") / Path(profile))
            file = self.resolve_include(path, relative_to)

        return file

    def resolve_base(self, base: str, relative_to: str) -> str | None:
        """Find the location of a base record in an extends directive.

        For a requested base b in path events/activity/thin.json, search:
          events/activity/b.json
          events/b.json
          b.json       # this should be ignored but isn't yet

        For a requested base b in path extn/stuff/events/activity/thing.json, search:
          extn/stuff/events/activity/b.json
          events/activity/b.json
          extn/stuff/events/b.json
          events/b.json
          extn/stuff/b.json
          b.json       # these last two should be ignored but aren't yet
          extn/b.json  #

        parameters:
            base: str         The base as it's described in the extends
                              directive, without a path or an extension.
            relative_to: str  The full path from the schema root to the record
                              extending the base.
        """
        base_path = Path(base)
        if base_path.suffix != ".json":
            base += ".json"

        # Search the current directory and each parent directory
        path = Path(relative_to)
        extn = self._types.extension(relative_to)

        while path != path.parent:
            test = str(path / base)
            if test in self._reader and test != relative_to:
                return test
            elif extn is not None:
                woextn = Path(*list(path.parts)[2:]) / base
                test = str(woextn)
                if test in self._reader:
                    return test

            path = path.parent

        return None

    def resolve_imprecise_base(self, base: str, relative_to: str) -> str | None:
        """Resolve an imprecise `extends` directive.

        Some `extends` directives point to files that are adjacent in the tree,
        e.g. ../blah/base.json. These seem imprecise to me without a strict
        lack of naming collisions, so I'm not making this the default behavior
        of `resolve_base()` in order to generate warnings.
        """
        base_path = Path(base)
        if base_path.suffix != ".json":
            base += ".json"

        # Search the current directory and each parent directory
        path = Path(relative_to)
        extn = self._types.extension(relative_to)

        while path != path.parent:
            for search in self._reader.ls(str(path)):
                search_path = path / search
                test = str(search_path / base)
                if test in self._reader and test != relative_to:
                    return test
                elif extn is not None:
                    woextn = Path(*list(search_path.parts)[2:]) / base
                    test = str(woextn)
                    if test in self._reader:
                        return test

            path = path.parent

        return None


class MergeParser:
    def __init__(
        self,
        reader: Reader,
        resolver: DependencyResolver,
        collector: Collector,
        types: TypeMapping,
    ):
        self._reader = reader
        self._resolver = resolver
        self._collector = collector
        self._types = types

    def applies_to(self, t: type) -> bool:
        return False

    def found_in(self, path: str) -> bool:
        return False

    def extract_targets(self, path: str) -> list[str]:
        return []

    def apply(self, path: str) -> None:
        for target in self.extract_targets(path):
            exclude = exclude_props(self._types[path], self._types[target])
            deep_merge(self._reader[path], self._reader[target], exclude=exclude)


class ExtendsParser(MergeParser):
    def applies_to(self, t: type) -> bool:
        if hasattr(t, "__required_keys__") or hasattr(t, "__optional_keys"):
            return EXTENDS_KEY in t.__required_keys__ or EXTENDS_KEY in t.__optional_keys__  # type: ignore
        else:
            return False

    def found_in(self, path: str) -> bool:
        return EXTENDS_KEY in self._reader[path]

    def extract_targets(self, path: str) -> list[str]:
        target = self._reader[path][EXTENDS_KEY]
        base = self._resolver.resolve_base(target, path)
        if base is None:
            base = self._resolver.resolve_imprecise_base(target, path)
            if base is None:
                self._collector.handle(ImpreciseBaseError(path, target))
            else:
                self._collector.handle(MissingBaseError(path, target))
        else:
            if self._types[base] not in [OcsfEvent, OcsfObject]:
                self._collector.handle(
                    IncludeTypeMismatchError(
                        path, base, "OcsfObject | OcsfEvent", "extends"
                    )
                )
            base = self._resolver.resolve_base(target, path)
            return [base if base is not None else ""]

        return []


class ProfilesParser(MergeParser):
    def applies_to(self, t: type) -> bool:
        if hasattr(t, "__required_keys__") or hasattr(t, "__optional_keys"):
            return PROFILES_KEY in t.__required_keys__ or PROFILES_KEY in t.__optional_keys__  # type: ignore

        else:
            return False

    def found_in(self, path: str) -> bool:
        return PROFILES_KEY in self._reader[path]

    def extract_targets(self, path: str) -> list[str]:
        targets = []
        profiles = self._reader[path][PROFILES_KEY]

        if isinstance(profiles, str):
            profiles = [profiles]

        for profile in profiles:
            target = self._resolver.resolve_profile(profile, path)
            if target is None:
                self._collector.handle(MissingProfileError(path, profile))
            else:
                targets.append(target)

        return targets


class AttributesParser(MergeParser):
    def applies_to(self, t: type) -> bool:
        if hasattr(t, "__required_keys__") or hasattr(t, "__optional_keys"):
            return ATTRIBUTES_KEY in t.__required_keys__ or ATTRIBUTES_KEY in t.__optional_keys__  # type: ignore
        else:
            return False

    def found_in(self, path: str) -> bool:
        return ATTRIBUTES_KEY in self._reader[path]

    def extract_targets(self, path: str) -> list[str]:
        if self._types[path] == OcsfDictionary:
            return []
        else:
            return [self._reader.key("dictionary.json")]
            # TODO the above should include extension dictionaries for correctness

    def _extn_dict(self, path):
        extn = self._types.extension(path)
        if extn is not None:
            dict_path = self._reader.key("extensions", extn, "dictionary.json")
            if dict_path in self._reader:
                return self._reader[dict_path][ATTRIBUTES_KEY]
        return {}

    def _root_dict(self):
        file = self._reader.find("dictionary.json")
        if file is not None:
            return file[ATTRIBUTES_KEY]
        return {}

    def apply(self, path: str):
        attrs = self._reader[path][ATTRIBUTES_KEY]
        root = self._root_dict()
        extn = self._extn_dict(path)

        # TODO is the dict name comparison enough or do we need to find by the `name` key?
        for name, attr in attrs.items():
            if name in extn:
                deep_merge(attrs[name], extn[name])
            if name in root:
                deep_merge(attrs[name], root[name])


class IncludeParser(MergeParser):
    def applies_to(self, t: type) -> bool:
        return ("__required_keys__" in t.__dict__ and INCLUDE_KEY in t.__required_keys__) or ("__optional_keys__" in t.__dict__ and INCLUDE_KEY in t.__optional_keys__)  # type: ignore

    def _has_includes(self, defn: dict[str, Any]) -> bool:
        """Recursively search for $include directives."""
        keys = list(defn.keys())
        for k in keys:
            if k == INCLUDE_KEY:
                return True
            elif isinstance(defn[k], dict):
                if self._has_includes(defn[k]):
                    return True
        return False

    def found_in(self, path: str) -> bool:
        return self._has_includes(self._reader[path])

    def _parse_includes(
        self,
        defn: dict[str, Any],
        path: str,
        trail: list[str] = [],
        update: bool = True,
        remove: bool = False,
    ) -> list[str]:
        """Find $include directives, optionally apply them, optionally
        remove the $include directive, and return a list of include targets.
        """
        keys = list(defn.keys())
        found = []

        for k in keys:
            if k == INCLUDE_KEY:
                if isinstance(defn[k], str):
                    targets = [defn[k]]
                else:
                    targets = defn[k]

                for target in targets:
                    t = self._resolver.resolve_include(target, path)
                    found.append(t)
                    if t is None:
                        self._collector.handle(MissingIncludeError(path, target))
                    elif update:
                        other = self._reader[t]
                        for key in trail:
                            other = other[key]
                        deep_merge(defn, other)

                if remove:
                    del defn[k]

            elif isinstance(defn[k], dict):
                found += self._parse_includes(
                    defn[k], path, trail + [k], update, remove
                )

        return found

    def extract_targets(self, path: str) -> list[str]:
        return self._parse_includes(
            self._reader[path], path, update=False, remove=False
        )

    def apply(self, path: str) -> None:
        self._parse_includes(self._reader[path], path, update=True, remove=False)


class Dependencies:
    """A friendly list of dependencies."""

    def __init__(self):
        self._dependencies: dict[str, list[tuple[str, str]]] = {}

    def add(self, child: str, parent: str, label: str = ""):
        if child not in self._dependencies:
            self._dependencies[child] = []
        self._dependencies[child].append((parent, label))

    def __iter__(self):
        return iter(self._dependencies)

    def __getitem__(self, key: str) -> list[tuple[str, str]]:
        return self._dependencies[key]

    def keys(self):
        return self._dependencies.keys()

    def exists(self, path: str, target: str, directive: Optional[str] = None):
        if path in self._dependencies:
            for item in self._dependencies[path]:
                if item[0] == target:
                    if directive is not None:
                        if directive == item[1]:
                            return True
                    else:
                        return True
        return False


def process_includes(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
    update: bool = True,
):
    if types is None:
        types = TypeMapping(reader, collector)

    resolver = DependencyResolver(reader, types)

    parsers = {
        EXTENDS_KEY: ExtendsParser(reader, resolver, collector, types),
        PROFILES_KEY: ProfilesParser(reader, resolver, collector, types),
        INCLUDE_KEY: IncludeParser(reader, resolver, collector, types),
        ATTRIBUTES_KEY: AttributesParser(reader, resolver, collector, types),
    }
    fulfilled: set[str] = set()
    dependencies = Dependencies()

    for path in reader.match():
        for directive, parser in parsers.items():
            if parser.found_in(path):
                for target in parser.extract_targets(path):
                    dependencies.add(path, target, directive)

    def process(path: str):
        if path not in fulfilled:
            if path in dependencies:
                for dependency, directive in dependencies[path]:
                    if dependency == path:
                        collector.handle(SelfInheritanceError(path, dependency))
                    elif directive == INCLUDE_KEY and dependencies.exists(
                        path, dependency, PROFILES_KEY
                    ):
                        collector.handle(RedundantProfileIncludeError(path, dependency))
                    else:
                        process(dependency)

            if update:
                for directive, parser in parsers.items():
                    if parser.found_in(path):
                        parser.apply(path)

            fulfilled.add(path)

    for path in dependencies.keys():
        # print(path, dependencies[path])
        process(path)

ocsf_validator/processor.py
query-ai/ocsf-validator
"""Tools for working with OCSF schema definition files.

This module contains tools to work with the JSON files that define the OCSF
schema. The most important export is the `Reader` class, which allows convenient
access to the OCSF schema as its represented in the definition files.
"""

import json
import re
from abc import ABC
from dataclasses import dataclass
from enum import IntEnum
from pathlib import Path, PurePath
from typing import Any, Callable, Dict, Iterable, Optional

from ocsf_validator.errors import InvalidBasePathError
from ocsf_validator.matchers import Matcher
from ocsf_validator.types import *

# TODO would os.PathLike be better?
Pathable = str | Path

# TODO refine Any in type signature
SchemaData = Dict[str, Any]

Pattern = str | Matcher


@dataclass
class ReaderOptions:
    """Options to control the behavior of a Reader."""

    base_path: Optional[Path] = None
    """The base path from which to load the schema."""

    read_extensions: bool = True
    """Recurse extensions."""


class Reader(ABC):
    """An in-memory copy of the raw OCSF schema definition.

    The `Reader` maintains a dictionary with relative file paths as strings
    and values of `Map[str, Any]` that are the decoded JSON of each schema
    file.

    Args:
        options (str | Path):    a base path for the schema, probably a clone of
                                 the `ocsf-schema` repository.
        options (ReaderOptions): an instances of ReaderOptions to change
                                 behaviors of the Reader.
    """

    def __init__(self, options: ReaderOptions | Pathable | None = None) -> None:
        if options is not None:
            if not isinstance(options, ReaderOptions):
                if isinstance(options, str):
                    path = Path(options)
                else:
                    path = options
                options = ReaderOptions(base_path=path)

            self._options = options
        else:
            self._options = ReaderOptions()

        self._data: SchemaData = {}
        self._root: str = ""

    def contents(self, path: Pathable) -> SchemaData:
        """Retrieve the parsed JSON data in a given file."""
        if isinstance(path, Path):
            path = str(path)

        # Can raise KeyError
        return self.__getitem__(path)

    def key(self, *args: str) -> str:
        """Platform agnostic key / filename creation from individual parts."""
        return str(self._root / Path(*args))

    def __getitem__(self, key: str):
        return self._data[key]

    def find(self, *parts: str) -> SchemaData | None:
        try:
            return self.__getitem__(self.key(*parts))
        except KeyError:
            return None

    def __setitem__(self, key: str, val: SchemaData):
        self._data[key] = val

    def __contains__(self, key: str):
        return key in self._data

    def __len__(self):
        return len(self._data)

    def ls(self, path: str | None = None, dirs=True, files=True) -> list[str]:
        if path is None:
            path = "/"
        if path[0] != "/":
            path = "/" + path

        base = Path(path)

        matched = set()
        for k in self._data.keys():
            p = Path(k)
            if p.parts[0 : len(base.parts)] == base.parts:
                depth = len(base.parts) + 1
                if (len(p.parts) == depth and files) or (len(p.parts) > depth and dirs):
                    matched.add(p.parts[len(base.parts)])

        return list(matched)

    def match(self, pattern: Optional[Pattern] = None) -> Iterable[str]:
        if pattern is not None:
            pattern = Matcher.make(pattern)

        for k in self._data.keys():
            if pattern is None or pattern.match(k):
                yield k

    def apply(self, op: Callable, pattern: Optional[Pattern] = None) -> None:
        """Apply a function to every 'file' in the schema, optionally if it
        matches a globbing expression `target`."""

        for k in self.match(pattern):
            op(self, k)

    def map(
        self,
        op: Callable,
        pattern: Optional[Pattern] = None,
        accumulator: Any = None,
    ) -> Any:
        """Apply a function to every 'file' in the schema, optionally if it
        matches a globbing expression `target`, and return the accumulated
        result."""

        for k in self.match(pattern):
            accumulator = op(self, k, accumulator)

        return accumulator


class DictReader(Reader):
    """A Reader that works from a `dict` without reading the filesystem.

    Useful (hopefully) for testing and debugging."""

    def set_data(self, data: SchemaData):
        self._data = data.copy()
        self._root = Path(next(iter(self._data.keys()))).root


class FileReader(Reader):
    """A Reader that reads schema definitions from JSON files."""

    def __init__(self, options: ReaderOptions | Pathable | None) -> None:
        if options is None:
            raise InvalidBasePathError("No base path specified")

        super().__init__(options)

        path = self._options.base_path

        if path is None:
            raise InvalidBasePathError(
                f"Missing schema base path in constructor arguments."
            )

        if not path.is_dir():
            raise InvalidBasePathError(f'Schema base path "{path}" is not a directory.')

        self._root = path.root
        self._data = _walk(path, path, self._options)


TRAVERSABLE_PATHS = ["enums", "includes", "objects", "events", "profiles", "extensions"]


def _walk(path: Path, base: Path, options: ReaderOptions) -> SchemaData:
    data: SchemaData = {}

    for entry in path.iterdir():
        key = str(base.root / entry.relative_to(base))

        if entry.is_file() and entry.suffix == ".json":
            with open(entry) as file:
                try:
                    data[key] = json.load(file)
                except json.JSONDecodeError as e:
                    # TODO maybe reformat this error before raising it
                    raise e

        elif entry.is_dir() and (
            entry.name in TRAVERSABLE_PATHS or entry.parent.name in TRAVERSABLE_PATHS
        ):
            if entry.name == "extensions" and not options.read_extensions:
                break

            data |= _walk(entry, base, options)

    return data

ocsf_validator/reader.py
query-ai/ocsf-validator
"""Validate OCSF Schema definitions.

"""

import traceback
from dataclasses import dataclass
from enum import IntEnum
from pathlib import Path
from typing import Callable, Optional

from termcolor import colored

import ocsf_validator.errors as errors
from ocsf_validator.processor import process_includes
from ocsf_validator.reader import FileReader, ReaderOptions
from ocsf_validator.type_mapping import TypeMapping
from ocsf_validator.validators import (validate_include_targets,
                                       validate_intra_type_collisions,
                                       validate_no_unknown_keys,
                                       validate_required_keys,
                                       validate_undefined_attrs,
                                       validate_unused_attrs)


class Severity(IntEnum):
    INFO = 0
    WARN = 1
    ERROR = 2
    FATAL = 3


@dataclass
class ValidatorOptions:
    """Configure validator behavior."""

    base_path: str = "."
    """The base path of the schema."""

    extensions: bool = True
    """Include the contents of extensions."""

    strict: bool = False
    """When True, exit with a non-zero exit code when warnings are encountered."""

    show_info: bool = False
    """Show informational messages."""

    invalid_path: int = Severity.FATAL
    """The OCSF Schema path could not be found or is horribly wrong."""

    invalid_metaschema: int = Severity.FATAL
    """The metaschema defined in this validator appears to be invalid."""

    missing_include: int = Severity.ERROR
    """An `$include` target is missing."""

    missing_profile: int = Severity.ERROR
    """A `profiles` target is missing."""

    missing_inheritance: int = Severity.ERROR
    """An `extends` inheritance target is missing."""

    imprecise_inheritance: int = Severity.WARN
    """An `extends` inheritance target is resolvable but imprecise and possibly ambiguous."""

    missing_key: int = Severity.ERROR
    """A required key is missing."""

    unknown_key: int = Severity.ERROR
    """An unrecognized key was found."""

    unused_attribute: int = Severity.WARN
    """An attribute in `dictionary.json` is unused."""

    self_inheritance: int = Severity.WARN
    """Attempting to `extend` the current record."""

    redundant_profile_include: int = Severity.INFO
    """Redundant profiles and $include target."""

    undetectable_type: int = Severity.WARN
    """Unable to detect type of file."""

    include_type_mismatch: int = Severity.WARN
    """Unexpected include type."""

    intra_type_name_collision: int = Severity.WARN
    """Same name used multiple times within a type."""

    undefined_attribute: int = Severity.WARN
    """Attributes used in a record but not defined in `dictionary.json`."""

    def severity(self, err: Exception):
        match type(err):
            case errors.MissingRequiredKeyError:
                return self.missing_key
            case errors.UnknownKeyError:
                return self.unknown_key
            case errors.MissingIncludeError:
                return self.missing_include
            case errors.MissingProfileError:
                return self.missing_profile
            case errors.MissingBaseError:
                return self.missing_inheritance
            case errors.UnusedAttributeError:
                return self.unused_attribute
            case errors.InvalidMetaSchemaError:
                return self.invalid_metaschema
            case errors.InvalidBasePathError:
                return self.invalid_path
            case errors.ImpreciseBaseError:
                return self.imprecise_inheritance
            case errors.SelfInheritanceError:
                return self.self_inheritance
            case errors.RedundantProfileIncludeError:
                return self.redundant_profile_include
            case errors.UndetectableTypeError:
                return self.undetectable_type
            case errors.IncludeTypeMismatchError:
                return self.include_type_mismatch
            case errors.TypeNameCollisionError:
                return self.intra_type_name_collision
            case errors.UndefinedAttributeError:
                return self.undefined_attribute
            case _:
                return Severity.INFO


class ValidationRunner:
    def __init__(self, pathOrOptions: str | ValidatorOptions):
        if isinstance(pathOrOptions, str):
            options = ValidatorOptions(base_path=pathOrOptions)
        else:
            options = pathOrOptions

        self.options = options

    def txt_fail(self, text: str):
        return colored(text, "red")

    def txt_warn(self, text: str):
        return colored(text, "yellow")

    def txt_crash(self, text: str):
        return colored(text, "black", "on_red")

    def txt_info(self, text: str):
        return colored(text, "blue")

    def txt_pass(self, text: str):
        return colored(text, "green")

    def txt_highlight(self, text: str):
        return colored(text, "light_grey", "on_cyan")

    def txt_emphasize(self, text: str):
        return colored(text, "white")

    def txt_label(self, severity: int):
        match severity:
            case Severity.INFO:
                return self.txt_info("INFO")
            case Severity.WARN:
                return self.txt_warn("WARNING")
            case Severity.ERROR:
                return self.txt_fail("ERROR")
            case Severity.FATAL:
                return self.txt_crash("FATAL")
            case _:
                return self.txt_emphasize("???")

    def validate(self):
        exit_code = 0
        messages: dict[str, dict[int, set[str]]] = {}
        collector = errors.Collector(throw=False)

        def test(label: str, code: Callable):
            message: str = ""
            code()

            if label not in messages:
                messages[label] = {}
                print("")
                print(self.txt_info("TESTING:"), self.txt_emphasize(label))

            for err in collector.exceptions():
                severity = self.options.severity(err)

                if severity not in messages[label]:
                    messages[label][severity] = set()

                messages[label][severity].add(str(err))
                if severity > Severity.INFO or self.options.show_info:
                    print("  ", self.txt_label(severity), err)

                if severity == Severity.FATAL:
                    exit(2)

            if len(collector) == 0:
                print("  ", self.txt_pass("PASS"), "No problems identified.")
            collector.flush()

        try:
            print(self.txt_emphasize("===[ OCSF Schema Validator ]==="))
            print(
                "Validating OCSF Schema at", self.txt_highlight(self.options.base_path)
            )

            # Setup the reader
            opts = ReaderOptions(
                base_path=Path(self.options.base_path),
                read_extensions=self.options.extensions,
            )
            reader = None
            try:
                reader = FileReader(opts)
            except errors.ValidationError as err:
                collector.handle(err)

            if reader is None:
                print(self.txt_crash("FATAL"), "Unable to initialize schema")
                exit()

            test("Schema definitions can be loaded", lambda: None)

            types = TypeMapping(reader, collector)
            test("Schema types can be inferred", lambda: None)

            # Validate dependencies
            test(
                "Dependency targets are resolvable and exist",
                lambda: validate_include_targets(
                    reader, collector=collector, types=types
                ),
            )

            process_includes(reader, collector=collector, types=types)

            # Any errors since the last test were duplicates; ignore them
            collector.flush()

            # Validate keys
            test(
                "Required keys are present",
                lambda: validate_required_keys(
                    reader, collector=collector, types=types
                ),
            )

            test(
                "There are no unrecognized keys",
                lambda: validate_no_unknown_keys(
                    reader, collector=collector, types=types
                ),
            )

            test(
                "All attributes in the dictionary are used",
                lambda: validate_unused_attrs(reader, collector=collector, types=types),
            )

            test(
                "All attributes are defined in dictionary.json",
                lambda: validate_undefined_attrs(
                    reader, collector=collector, types=types
                ),
            )

            test(
                "Names are not used multiple times within a record type",
                lambda: validate_intra_type_collisions(
                    reader, collector=collector, types=types
                ),
            )

        except Exception as err:
            print("Encountered an unexpected exception:")
            traceback.print_exception(err)

        finally:
            print("")
            print(self.txt_emphasize("SUMMARY"))

            failure_threshold = (
                Severity.ERROR if not self.options.strict else Severity.WARN
            )

            for k in messages:
                found = False
                if len(messages[k].items()) > 0:
                    for sev in [
                        Severity.FATAL,
                        Severity.ERROR,
                        Severity.WARN,
                        Severity.INFO,
                    ]:
                        if sev in messages[k] and sev >= failure_threshold:
                            found = True
                            print("  ", self.txt_fail("FAILED"), k)
                            exit_code = 1

                if not found:
                    print("  ", self.txt_pass("PASSED"), k)

            print("")
            exit(exit_code)

ocsf_validator/runner.py
query-ai/ocsf-validator
from pathlib import Path

from ocsf_validator.errors import Collector, UndetectableTypeError
from ocsf_validator.matchers import *
from ocsf_validator.reader import Reader
from ocsf_validator.types import *

MATCHERS = [
    VersionMatcher(),
    DictionaryMatcher(),
    CategoriesMatcher(),
    IncludeMatcher(),
    CategoriesMatcher(),
    ProfileMatcher(),
    ObjectMatcher(),
    EventMatcher(),
    ExtensionMatcher(),
]


class TypeMapping:
    def __init__(self, reader: Reader, collector: Collector = Collector.default):
        self._reader = reader
        self._collector = collector
        self._mappings: dict[str, type] = {}
        self.update()

    def __getitem__(self, path: str) -> type:
        return self._mappings[path]

    def __contains__(self, path: str) -> bool:
        return path in self._mappings

    def __iter__(self):
        return iter(self._mappings)

    def _get_type(self, path: str) -> type | None:
        for matcher in MATCHERS:
            if matcher.match(path):
                return matcher.get_type()
        return None

    def update(self):
        for path in self._reader.match():
            t = self._get_type(path)
            if t is not None:
                self._mappings[path] = t
            else:
                self._collector.handle(UndetectableTypeError(path))

    def extension(self, path: str) -> str | None:
        """Extract the extension name from a given key/filepath."""
        parts = list(Path(self._reader.key(path)).parts)
        if "extensions" in parts:
            return parts[parts.index("extensions") + 1]
        else:
            return None

ocsf_validator/type_mapping.py
query-ai/ocsf-validator
from dataclasses import dataclass, field
from typing import (Any, Dict, NotRequired, Optional, Required, Sequence,
                    TypedDict, TypeVar, Union)

ATTRIBUTES_KEY = "attributes"
PROFILES_KEY = "profiles"
EXTENDS_KEY = "extends"
INCLUDE_KEY = "$include"


class OcsfVersion(TypedDict):
    version: str


class OcsfEnumMember(TypedDict):
    caption: str
    description: NotRequired[str]
    notes: NotRequired[str]


class OcsfEnum(TypedDict):
    enum: Dict[str, OcsfEnumMember]


class OcsfDeprecationInfo(TypedDict):
    message: Required[str]
    since: Required[str]


OcsfAttr = TypedDict(
    "OcsfAttr",
    {
        "$include": NotRequired[str],
        # "caption": NotRequired[str],
        "caption": str,
        "default": NotRequired[Any],
        "description": NotRequired[str],
        "enum": NotRequired[Dict[str, OcsfEnumMember]],
        "group": NotRequired[str],
        "is_array": NotRequired[bool],
        "max_len": NotRequired[int],
        "name": NotRequired[str],
        "notes": NotRequired[str],
        "observable": NotRequired[int],
        "range": NotRequired[Sequence[int]],
        "regex": NotRequired[str],
        "requirement": NotRequired[str],
        "sibling": NotRequired[str],
        "type": NotRequired[str],
        "type_name": NotRequired[str],
        "profile": NotRequired[Optional[Sequence[str]]],
        "values": NotRequired[Sequence[Any]],
        "@deprecated": NotRequired[OcsfDeprecationInfo],
    },
)


class OcsfExtension(TypedDict):
    uid: int
    name: str
    caption: str
    path: NotRequired[str]
    version: NotRequired[str]
    description: NotRequired[str]


class OcsfDictionaryTypes(TypedDict):
    attributes: Dict[str, OcsfAttr]
    caption: str
    description: str


class OcsfDictionary(TypedDict):
    attributes: Dict[str, OcsfAttr]
    caption: str
    description: str
    name: str
    types: NotRequired[OcsfDictionaryTypes]


class OcsfCategory(TypedDict):
    caption: str
    description: str
    uid: int


class OcsfCategories(TypedDict):
    attributes: Dict[str, OcsfCategory]
    caption: str
    description: str
    name: str


class OcsfInclude(TypedDict):
    caption: str
    attributes: Dict[str, OcsfAttr]
    description: NotRequired[str]
    annotations: NotRequired[Dict[str, str]]


class OcsfProfile(TypedDict):
    caption: str
    caption: str
    description: str
    meta: str
    name: str
    attributes: Dict[str, OcsfAttr]
    annotations: NotRequired[Dict[str, str]]


OcsfObject = TypedDict(
    "OcsfObject",
    {
        "caption": str,
        "description": str,
        "name": str,
        "attributes": Dict[str, OcsfAttr],
        "extends": NotRequired[Union[str, list[Optional[str]]]],
        "observable": NotRequired[int],
        "profiles": NotRequired[Sequence[str]],
        "constraints": NotRequired[Dict[str, Sequence[str]]],
        "$include": NotRequired[Union[str, Sequence[str]]],
        "@deprecated": NotRequired[OcsfDeprecationInfo],
    },
)


OcsfEvent = TypedDict(
    "OcsfEvent",
    {
        "attributes": Dict[str, OcsfAttr],
        "caption": str,
        "name": str,
        "uid": NotRequired[int],
        "category": NotRequired[str],
        "description": NotRequired[str],
        "extends": NotRequired[Union[str, list[Optional[str]]]],
        "profiles": NotRequired[Sequence[str]],
        "associations": NotRequired[Dict[str, Sequence[str]]],
        "constraints": NotRequired[Dict[str, Sequence[str]]],
        "$include": NotRequired[Union[str, Sequence[str]]],
        "@deprecated": NotRequired[OcsfDeprecationInfo],
    },
)

T = TypeVar("T")
PerExt = Dict[Optional[str], T]
# Includable = Union[OcsfInclude, OcsfEnum, OcsfProfile]


class OcsfSchema(TypedDict):
    categories: PerExt[OcsfCategories]
    dictionaries: PerExt[OcsfDictionary]
    observables: Dict[int, OcsfObject | OcsfAttr]
    objects: PerExt[Dict[str, OcsfObject]]
    events: PerExt[Dict[str, OcsfEvent]]
    includes: PerExt[Dict[str, OcsfInclude]]
    profiles: PerExt[Dict[str, OcsfProfile]]
    enums: PerExt[Dict[str, OcsfEnum]]


def is_ocsf_type(t: type):
    return (
        t is OcsfEnumMember
        or t is OcsfEnum
        or t is OcsfDeprecationInfo
        or t is OcsfAttr
        or t is OcsfExtension
        or t is OcsfDictionaryTypes
        or t is OcsfDictionary
        or t is OcsfCategory
        or t is OcsfCategories
        or t is OcsfInclude
        or t is OcsfProfile
        or t is OcsfObject
        or t is OcsfEvent
    )


def leaf_type(defn: type, prop: str) -> type | None:
    if hasattr(defn, "__annotations__") and prop in defn.__annotations__:
        t = defn.__annotations__[prop]
        if hasattr(t, "__args__"):
            return t.__args__[-1]
        else:
            return t
    return None

ocsf_validator/types.py
query-ai/ocsf-validator
import inspect
from typing import (Dict, NotRequired, Optional, Required, get_type_hints,
                    is_typeddict)

from ocsf_validator.errors import (Collector, InvalidMetaSchemaError,
                                   MissingRequiredKeyError,
                                   TypeNameCollisionError,
                                   UndefinedAttributeError,
                                   UndetectableTypeError, UnknownKeyError,
                                   UnusedAttributeError)
from ocsf_validator.matchers import (AnyMatcher, DictionaryMatcher,
                                     EventMatcher, ExtensionMatcher,
                                     IncludeMatcher, ObjectMatcher,
                                     ProfileMatcher)
from ocsf_validator.processor import process_includes
from ocsf_validator.reader import Reader
from ocsf_validator.type_mapping import TypeMapping
from ocsf_validator.types import *


def validate_required_keys(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
):
    """Validate that no required keys are missing."""

    if types is None:
        types = TypeMapping(reader)

    def compare_keys(
        data: Dict[str, Any], defn: type, file: str, trail: list[str] = []
    ):
        if hasattr(defn, "__required_keys__"):
            for k in defn.__required_keys__:  # type: ignore
                t = leaf_type(defn, k)
                if k not in data:
                    collector.handle(MissingRequiredKeyError(k, file, defn, trail))
                elif t is not None and is_ocsf_type(t):
                    if isinstance(data[k], dict):
                        # dict[str, Ocsf____]
                        for k2, val in data[k].items():
                            if k2 != INCLUDE_KEY:
                                compare_keys(data[k][k2], t, file, trail + [k, k2])
                    else:
                        compare_keys(data[k], t, file, trail + [k])

        else:
            collector.handle(
                InvalidMetaSchemaError(
                    f"Unexpected definition {defn} used when processing {file}"
                )
            )

    def validate(reader: Reader, file: str):
        record = reader[file]
        if file not in types:
            collector.handle(UndetectableTypeError(file))
        else:
            defn = types[file]
            if not hasattr(defn, "__annotations__"):
                collector.handle(InvalidMetaSchemaError(f"{defn} is not a TypedDict"))
            compare_keys(record, defn, file)

    reader.apply(validate)


def validate_no_unknown_keys(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
):
    """Validate that there are no unknown keys."""

    if types is None:
        types = TypeMapping(reader)

    def compare_keys(
        data: Dict[str, Any], defn: type, file: str, trail: list[str] = []
    ):
        if hasattr(defn, "__annotations__") and isinstance(data, dict):
            for k in data.keys():
                t = leaf_type(defn, k)
                if t is None:
                    collector.handle(UnknownKeyError(k, file, defn, trail))
                elif is_ocsf_type(t):
                    if hasattr(defn.__annotations__[k], "__args__"):
                        args = defn.__annotations__[k].__args__
                        if len(args) >= 2:
                            if args[-2] == str:
                                for k2, val in data[k].items():
                                    if k2 != INCLUDE_KEY:
                                        compare_keys(
                                            data[k][k2], t, file, trail + [k, k2]
                                        )
                            else:
                                ...  # what would this be?
                        else:
                            compare_keys(data[k], args[-1], file, trail + [k])
                    else:
                        compare_keys(data[k], t, file, trail + [k])

        else:
            collector.handle(
                InvalidMetaSchemaError(
                    f"Unexpected definition {defn} used when processing {file}"
                )
            )

    def validate(reader: Reader, file: str):
        record = reader[file]
        if file not in types:
            collector.handle(UndetectableTypeError(file))
        else:
            defn = types[file]
            if not hasattr(defn, "__annotations__"):
                collector.handle(InvalidMetaSchemaError(f"{defn} is not a TypedDict"))
            compare_keys(record, defn, file)

    reader.apply(validate)


def validate_include_targets(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
):
    process_includes(reader, collector=collector, types=types, update=False)


def validate_unused_attrs(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
):
    if types is None:
        types = TypeMapping(reader)

    # TODO: Lift validate() function out and use a TypeMapping
    def make_validator(defn: type):
        def validate(reader: Reader, key: str, accum: set[str]):
            record = reader[key]
            if ATTRIBUTES_KEY in record:
                return accum | set(
                    [k for k in record[ATTRIBUTES_KEY]]
                )  # should it be defn[attrs][k]['name'] ?
            else:
                return accum

        return validate

    attrs = reader.map(make_validator(OcsfObject), ObjectMatcher(), set())
    attrs |= reader.map(make_validator(OcsfEvent), EventMatcher(), set())

    d = reader.find("dictionary.json")

    if d is not None:
        for k in d[ATTRIBUTES_KEY]:
            if k not in attrs:
                collector.handle(UnusedAttributeError(k))


def validate_undefined_attrs(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
):
    if types is None:
        types = TypeMapping(reader)

    EXCLUDE = ["$include"]

    def validate(reader: Reader, file: str):
        record = reader[file]
        if ATTRIBUTES_KEY in record:
            d = reader.find("dictionary.json")
            if d is not None:
                for k in record[ATTRIBUTES_KEY]:
                    if k not in d[ATTRIBUTES_KEY] and k not in EXCLUDE:
                        collector.handle(UndefinedAttributeError(k, file))
            else:
                collector.handle(InvalidMetaSchemaError())

    reader.apply(
        validate,
        AnyMatcher(
            [ObjectMatcher(), EventMatcher(), ProfileMatcher(), IncludeMatcher()]
        ),
    )


def validate_intra_type_collisions(
    reader: Reader,
    collector: Collector = Collector.default,
    types: Optional[TypeMapping] = None,
):
    if types is None:
        types = TypeMapping(reader)

    found: dict[str, dict[str, list[str]]] = {}

    def validate(reader: Reader, file: str):
        t = str(types[file])
        if t not in found:
            found[t] = {}

        if "name" in reader[file]:
            name = reader[file]["name"]
            if name not in found[t]:
                found[t][name] = []
            else:
                collector.handle(TypeNameCollisionError(name, t, file, found[t][name][0]))
            found[t][name].append(file)

    reader.apply(validate, AnyMatcher([ObjectMatcher(), EventMatcher()]))

ocsf_validator/validators.py
query-ai/ocsf-validator
import pytest

from ocsf_validator.errors import *
from ocsf_validator.processor import *
from ocsf_validator.reader import DictReader, Reader


def attributes(attrs=[]):
    d = {}
    for a in attrs:
        d[a] = {"name": a}
    return {"attributes": d}


def obj(name="object", attrs=[]):
    return {"name": name, "caption": ""} | attributes(attrs)


def event(name="event", attrs=[]):
    return {"name": name, "caption": ""} | attributes(attrs)


def test_include_one():
    net = attributes(["proxy", "src_ip"])
    net["name"] = "network"
    net["name2"] = "network"
    httpa = event("http_activity")
    httpa["$include"] = "includes/network.json"

    s = {
        "/events/network/http_activity.json": httpa,
        "/includes/network.json": net,
        "/dictionary.json": attributes(["stuff"]),
    }

    r = DictReader()
    r.set_data(s)
    process_includes(r)

    assert "attributes" in r["/events/network/http_activity.json"]
    assert "name2" in r["/events/network/http_activity.json"]
    assert r["/events/network/http_activity.json"]["name"] == "http_activity"
    assert "proxy" in r["/events/network/http_activity.json"]["attributes"]


def test_include_many():
    net = attributes(["proxy", "src_ip"])
    thing = attributes(["dest_ip", "score"])
    httpa = event("http_activity")
    httpa["$include"] = ["includes/network.json", "events/thing.json"]

    s = {
        "/events/network/http_activity.json": httpa,
        "/includes/network.json": net,
        "/events/thing.json": thing,
        "/dictionary.json": attributes(["stuff"]),
    }

    r = DictReader()
    r.set_data(s)
    process_includes(r)

    assert "attributes" in r["/events/network/http_activity.json"]
    assert r["/events/network/http_activity.json"]["name"] == "http_activity"
    assert "proxy" in r["/events/network/http_activity.json"]["attributes"]
    assert "dest_ip" in r["/events/network/http_activity.json"]["attributes"]


def test_include_attrs():
    net = attributes(["proxy", "src_ip"])
    thing = attributes(["dest_ip", "score"])
    httpa = event("http_activity")
    httpa["attributes"]["$include"] = ["includes/network.json", "events/thing.json"]

    s = {
        "/events/network/http_activity.json": httpa,
        "/includes/network.json": net,
        "/events/thing.json": thing,
        "/dictionary.json": attributes(["stuff"]),
    }

    r = DictReader()
    r.set_data(s)
    process_includes(r)

    assert "attributes" in r["/events/network/http_activity.json"]
    assert r["/events/network/http_activity.json"]["name"] == "http_activity"
    assert "proxy" in r["/events/network/http_activity.json"]["attributes"]
    assert "dest_ip" in r["/events/network/http_activity.json"]["attributes"]


def test_missing_include():
    httpa = event("http_activity")
    httpa["attributes"]["$include"] = "includes/network.json"

    s = {
        "/events/network/http_activity.json": httpa,
        "/dictionary.json": attributes(["stuff"]),
    }

    r = DictReader()
    r.set_data(s)

    with pytest.raises(MissingIncludeError):
        process_includes(r)


def test_extends():
    base = event("base_event", ["thing"])
    httpa = event("http_activity")
    httpa["extends"] = "base_event"

    s = {
        "/events/network/http_activity.json": httpa,
        "/events/base_event.json": base,
        "/dictionary.json": attributes(["stuff"]),
    }
    r = DictReader()
    r.set_data(s)

    process_includes(r)

    assert "thing" in r["/events/network/http_activity.json"]["attributes"]


def test_profiles():
    prof = event("profile1", ["thing"])
    httpa = event("http_activity")
    httpa["profiles"] = "profile1"

    s = {
        "/events/network/http_activity.json": httpa,
        "/profiles/profile1.json": prof,
        "/dictionary.json": attributes(["stuff"]),
    }
    r = DictReader()
    r.set_data(s)

    process_includes(r)

    assert "thing" in r["/events/network/http_activity.json"]["attributes"]


def test_profiles_many():
    prof1 = event("profile1", ["thing1"])
    prof2 = event("profile2", ["thing2"])
    httpa = event("http_activity")
    httpa["profiles"] = ["profile1", "profile2"]

    s = {
        "/events/network/http_activity.json": httpa,
        "/profiles/profile1.json": prof1,
        "/profiles/profile2.json": prof2,
        "/dictionary.json": attributes(["stuff"]),
    }
    r = DictReader()
    r.set_data(s)

    process_includes(r)

    assert "thing1" in r["/events/network/http_activity.json"]["attributes"]
    assert "thing2" in r["/events/network/http_activity.json"]["attributes"]


def test_profiles():
    prof = event("profile1", ["thing"])
    prof["meta"] = "stuff"
    httpa = event("http_activity")
    httpa["profiles"] = "profile1"
    prof2 = event("profile1", ["thing2"])
    neta = event("network_activity")
    neta["profiles"] = "profile1"

    s = {
        "/extensions/one/events/network/http_activity.json": httpa,
        "/extensions/one/profiles/profile1.json": prof,
        "/events/network/net_activity.json": neta,
        "/profiles/profile1.json": prof2,
        "/dictionary.json": attributes(["stuff"]),
    }
    r = DictReader()
    r.set_data(s)

    process_includes(r)

    assert (
        "thing" in r["/extensions/one/events/network/http_activity.json"]["attributes"]
    )
    assert "meta" not in r["/extensions/one/events/network/http_activity.json"]
    assert (
        "thing2"
        not in r["/extensions/one/events/network/http_activity.json"]["attributes"]
    )
    assert "thing" not in r["/events/network/net_activity.json"]["attributes"]
    assert "thing2" in r["/events/network/net_activity.json"]["attributes"]


def test_attrs_from_dictionary():
    o1 = obj("o1", ["thing"])
    o1["attributes"]["thing"]["name"] = "thing1"

    d = {
        "attributes": {
            "thing": {"name": "thing", "caption": "Thing", "requirement": "optional"},
            "thing2": {"name": "thing2"},
        }
    }

    s = {
        "/objects/o1.json": o1,
        "/dictionary.json": d,
    }
    r = DictReader()
    r.set_data(s)

    process_includes(r)
    assert "thing" in r["/objects/o1.json"]["attributes"]
    assert r["/objects/o1.json"]["attributes"]["thing"]["name"] is "thing1"
    assert "thing2" not in r["/objects/o1.json"]["attributes"]
    assert "requirement" in r["/objects/o1.json"]["attributes"]["thing"]

tests/test_dependencies.py
query-ai/ocsf-validator
import pytest

from ocsf_validator.reader import DictReader
from ocsf_validator.type_mapping import *


def test_mapping():
    s = {
        "/dictionary.json": {},
        "/objects/object.json": {},
        "/categories.json": {},
        "/profiles/profile.json": {},
        "/version.json": {},
        "/events/event.json": {},
        "/extensions/a/events/event.json": {},
        "/extensions/a/objects/object.json": {},
        "/extensions/a/profiles/profile.json": {},
    }
    r = DictReader()
    r.set_data(s)
    tm = TypeMapping(r)

    assert tm["/dictionary.json"] is OcsfDictionary
    assert tm["/events/event.json"] is OcsfEvent
    assert tm["/extensions/a/events/event.json"] is OcsfEvent
    assert tm["/objects/object.json"] is OcsfObject
    assert tm["/extensions/a/objects/object.json"] is OcsfObject
    assert tm["/categories.json"] is OcsfCategories
    assert tm["/version.json"] is OcsfVersion
    assert tm["/profiles/profile.json"] is OcsfProfile
    assert tm["/extensions/a/profiles/profile.json"] is OcsfProfile

tests/test_mapping.py
query-ai/ocsf-validator
import pytest

from ocsf_validator.matchers import *
from ocsf_validator.types import *


def test_dictionary_matcher():
    m = DictionaryMatcher()

    assert m.match("dictionary.json") is True
    assert m.match("/dictionary.json") is True
    assert m.match("/extension/win/dictionary.json") is True
    assert m.match("/objects/thing.json") is False
    assert m.get_type() is OcsfDictionary


def test_object_matcher():
    m = ObjectMatcher()

    assert m.match("/objects/thing.json") is True
    assert m.match("/extensions/win/objects/thing.json") is True
    assert m.match("objects/win/objects/thing.json") is True
    assert m.match("/events/thing.json") is False
    assert m.get_type() is OcsfObject


def test_event_matcher():
    m = EventMatcher()

    assert m.match("/events/base_event.json") is True
    assert m.match("events/activity/network_activity.json") is True
    assert m.match("events/filesystem/filesystem.json") is True
    assert m.match("/extensions/win/events/activity/network_activity.json") is True
    assert m.get_type() is OcsfEvent


def test_extension_matcher():
    m = ExtensionMatcher()

    assert m.match("/extensions/ext1/extension.json") is True
    assert m.match("/extension.json") is False
    assert m.get_type() is OcsfExtension


def test_make_matcher():
    m = Matcher.make(".*thing.json")

    assert m.match("thing.json") is True

tests/test_matchers.py
query-ai/ocsf-validator
from pathlib import Path

import pytest

from ocsf_validator.matchers import GlobMatcher
from ocsf_validator.reader import DictReader, Reader

event = {"name": "an event"}
obj = {"name": "an object"}
data = {
    "/events/base_event.json": event.copy(),
    "/events/application/application.json": event.copy(),
    "/objects/os.json": obj.copy(),
    "/extensions/win/objects/win_process.json": obj.copy(),
    "/extensions/win/events/system/registry_key.json": event.copy(),
    "/dictionary.json": obj.copy(),
}


def reader():
    r = DictReader()
    r.set_data(data)
    return r


def test_get_item():
    r = reader()
    assert r["/objects/os.json"] == obj
    assert r.contents("/objects/os.json") == obj


def test_set_item():
    r = reader()
    r["/objects/api.json"] = {"name": "api"}
    assert r["/objects/api.json"]["name"] == "api"


def test_apply():
    r = reader()

    def mark(reader: Reader, key: str):
        reader[key]["test"] = True

    r.apply(mark, GlobMatcher("objects/*"))
    assert r["/objects/os.json"]["test"] == True
    assert r["/extensions/win/objects/win_process.json"]["test"] == True


def test_find():
    r = reader()
    f = r.find("objects", "os.json")
    assert "name" in f


def test_map():
    r = reader()

    def f(reader: Reader, key: str, acc: int):
        return acc + 1

    matches = r.map(f, GlobMatcher("objects/*"), 0)
    assert matches == 2


def test_ls():
    r = reader()

    matches = r.ls()
    assert "objects" in matches
    assert "win" not in matches

    matches = r.ls("objects")
    assert "os.json" in matches
    assert "win" not in matches

    matches = r.ls("extensions")
    assert "os.json" not in matches
    assert "win" in matches

    matches = r.ls("events", files=False)
    assert "application" in matches
    assert "base_event.json" not in matches

    matches = r.ls("events", dirs=False)
    assert "application" not in matches
    assert "base_event.json" in matches

tests/test_reader.py
query-ai/ocsf-validator
import pytest

from ocsf_validator.types import *


def test_is_ocsf_type():
    assert is_ocsf_type(OcsfDictionary) is True
    assert is_ocsf_type(OcsfAttr) is True
    assert is_ocsf_type(str) is False

tests/test_types.py
query-ai/ocsf-validator
import pytest

from ocsf_validator.errors import *
from ocsf_validator.reader import DictReader
from ocsf_validator.validators import *

d1 = {
    "/extensions/ext1/extension.json": {
        "uid": 1,
        "name": "ext1",
        "path": "ext1",
        # "caption": "Extension One"
        "color": "blue",
    }
}


def test_required_keys():
    r = DictReader()
    r.set_data(d1)

    with pytest.raises(MissingRequiredKeyError):
        validate_required_keys(r)


def test_deep_required_keys():
    s = {
        "/events/event.json": {
            "caption": "Event",
            "name": "event",
            "attributes": {
                "one": {
                    "name": "one",
                },
            },
        },
    }
    r = DictReader()
    r.set_data(s)

    with pytest.raises(MissingRequiredKeyError) as exc:
        validate_required_keys(r)
    assert exc.value.key is "caption"


def test_unknown_keys():
    r = DictReader()
    r.set_data(d1)

    with pytest.raises(UnknownKeyError):
        validate_no_unknown_keys(r)


def test_validate_unused_attrs():
    r = DictReader()
    r.set_data(
        {
            "/dictionary.json": {
                "attributes": {
                    "one": {
                        "name": "one",
                        "caption": "One",
                    },
                    "two": {
                        "name": "two",
                        "caption": "Two",
                    },
                    "three": {
                        "name": "three",
                        "caption": "Three",
                    },
                },
            },
            "/objects/thing.json": {
                "name": "thing",
                "attributes": {
                    "one": {"name": "one"},
                },
            },
            "/events/stuff/another-thing.json": {
                "name": "thing",
                "attributes": {
                    "two": {"name": "two"},
                },
            },
        }
    )

    with pytest.raises(UnusedAttributeError) as exc:
        validate_unused_attrs(r)
    assert exc.value.attr == "three"


def test_validate_undefined_attrs():
    r = DictReader()
    r.set_data(
        {
            "/dictionary.json": {
                "attributes": {
                    "one": {
                        "name": "one",
                        "caption": "One",
                    },
                },
            },
            "/objects/thing.json": {
                "name": "thing",
                "attributes": {
                    "one": {"name": "one"},
                    "two": {"name": "two"},
                },
            },
        }
    )

    with pytest.raises(UndefinedAttributeError) as exc:
        validate_undefined_attrs(r)
    assert exc.value.attr == "two"


def test_validate_intra_type_collisions():
    r = DictReader()
    r.set_data(
        {
            "/objects/thing.json": {
                "name": "thing",
                "attributes": {
                    "one": {"name": "one"},
                    "two": {"name": "two"},
                },
            },
            "/objects/thing2.json": {
                "name": "thing",
                "attributes": {},
            },
        }
    )

    with pytest.raises(TypeNameCollisionError) as exc:
        validate_intra_type_collisions(r)
    assert exc.value.name == "thing"

    r["/events/event.json"] = {"name": "thing"}
    r["/objects/thing2.json"] = {"name": "thing2"}
    # no error
    validate_intra_type_collisions(r)

tests/test_validators.py
query-ai/ocsf-validator