assets_guardian.core.domain.engines.compliance_engine

Classes

ComplianceEngine(rules)

IAM compliance engine.

class assets_guardian.core.domain.engines.compliance_engine.ComplianceEngine(rules: list[IRule])

Bases: object

IAM compliance engine.

Responsible for executing audit rules by category (comparison, compliance, matrix) on the collected data.

Variables:

_rules – List of rule instances to evaluate.

Initializes the engine with a list of rules.

Parameters:

rules – List of rule instances to evaluate.

run_all(*, old_data: Mapping[str, Iterable[Any]] | None = None, new_data: Mapping[str, Iterable[Any]] | None = None, live_data: Mapping[str, Iterable[Any]] | None = None, config: dict[str, Any] | None = None, accesses: Iterable[Any] | None = None, matrix: dict[tuple[str, str], str] | None = None, profiles: dict[str, list[str]] | None = None) → Iterator[Finding]

Executes all rule categories and yields findings incrementally.

Parameters:

• old_data – Comparison data (old state).

• new_data – Comparison data (new state).

• live_data – Data for standard compliance.

• config – Global rules configuration.

• accesses – Actual access data for matrix checks.

• matrix – Authorized rights matrix.

• profiles – User profiles mapping.

Yields:

Finding – All anomalies detected by all rules.

run_comparison(old_data: Mapping[str, Iterable[Any]], new_data: Mapping[str, Iterable[Any]]) → Iterator[Finding]

Executes comparison rules in streaming mode.

Parameters:

• old_data – Data extracted from the old state (e.g., previous Excel sheet).

• new_data – Freshly collected data.

Yields:

Finding – Detected discrepancies on the fly.

run_compliance(live_data: Mapping[str, Iterable[Any]], config: dict[str, Any]) → Iterator[Finding]

Executes IAM compliance rules in streaming mode.

Parameters:

• live_data – Data collected in real time.

• config – Configuration parameters (thresholds, etc.).

Yields:

Finding – Detected non-conformities on the fly.

run_matrix(accesses: Iterable[Any], matrix: dict[tuple[str, str], str], profiles: dict[str, list[str]]) → Iterator[Finding]

Executes matrix checks in streaming mode.

Parameters:

• accesses – List of actual collected accesses.

• matrix – Dictionary representing the matrix (profile -> asset -> role).

• profiles – Mapping of employees to their expected profiles.

Yields:

Finding – Detected unauthorized or missing accesses.