Feature: Policy Gating

Relevant source files

The following files were used as context for generating this wiki page:

• core/policy.py
• core/configuration.py
• config/policy.yml
• core/stage_runner.py
• tests/test_policy.py
• core/connectors.py
• tests/test_evidence_bundle.py

The Policy Gating feature in Aldeci provides a framework for enforcing rules and making automated decisions based on security and compliance data. It integrates with an Open Policy Agent (OPA) server to evaluate input data against a defined set of policies. Based on the outcome, it can trigger various actions, such as logging warnings, failing a process, or dispatching notifications to external systems like Jira and Slack.

This feature is central to implementing security-as-code, allowing teams to define guardrails and automated responses within a configuration-driven pipeline. The gating mechanism is used in various stages of the SSDLC, most notably in the decision stage, to determine whether a release should be allowed, deferred, or blocked based on aggregated risk factors and control failures.

Architecture

The Policy Gating system is composed of three main parts: the Policy Configuration, the OPA Client, and the Policy Automation engine. These components work together to evaluate data and execute actions.

Sources: core/policy.py, core/configuration.py

Core Components

The following diagram shows the main classes involved in the policy gating feature and their relationships.

*PolicyAutomation**: The primary orchestrator. It takes an input document, sends it to the OPA client for evaluation, and then uses the dispatcher to execute actions based on the policy result.
* Sources: core/policy.py:38-66

• _OPAClient: A simple HTTP client responsible for sending input data to an external OPA server and returning the decision.
  • Sources: core/policy.py:13-36
• _AutomationDispatcher: Executes the actions defined in the policy, such as creating a Jira ticket or sending a Slack message, using the configured connectors.
  • Sources: core/policy.py:69-87
• AutomationConnectors: A container for clients to external systems like Jira, Slack, and Confluence.
  • Sources: core/connectors.py:100-120

Policy Configuration

Policy behavior is defined in a YAML configuration file, typically config/policy.yml, which is loaded and validated at runtime. The configuration is structured into engine, actions, and triggers.

Sources: core/configuration.py:125-168, config/policy.yml

Configuration Structure

The policy configuration is validated by the _validate_policy_config function.

# Example from config/policy.yml
policy:
  engine:
    # OPA server URL
    url: http://localhost:8181/v1/data/fixops/allow
    # Request timeout in seconds
    timeout_seconds: 5
  actions:
    # Defines what to do for each action type
    warn:
      - handler: log
        params:
          message: "Policy warning triggered"
    fail:
      - handler: log
        params:
          message: "Policy failure triggered"
  triggers:
    # Maps policy results to actions
    on_warn:
      - warn
    on_fail:
      - fail

Sources: config/policy.yml:1-18, core/configuration.py:138-151

Key Configuration Options

Section	Key	Type	Description	Source
engine	url	string	The URL of the OPA server endpoint to query for decisions.	core/configuration.py:165-166
engine	timeout_seconds	integer	The timeout in seconds for requests to the OPA server.	core/configuration.py:167-168
actions	(dynamic)	list	A mapping of action names (e.g., warn) to a list of handlers to execute.	core/configuration.py:152-155
triggers	on_warn	list	A list of actions from the actions section to run when the policy result is "warn".	core/configuration.py:139-143
triggers	on_fail	list	A list of actions from the actions section to run when the policy result is "fail".	core/configuration.py:139-143
triggers	on_defer	list	A list of actions from the actions section to run when the policy result is "defer".	core/configuration.py:139-143

Evaluation Flow

The policy evaluation process is initiated by calling the evaluate method on a PolicyAutomation instance. The flow involves querying the OPA server and dispatching actions based on the response.

This sequence diagram illustrates the interaction between components during a policy evaluation.

1. An external caller (like a CI script or API endpoint) invokes PolicyAutomation.evaluate() with an input document and a trigger (e.g., on_fail).
2. PolicyAutomation passes the input document to the _OPAClient.
3. _OPAClient sends an HTTP POST request to the configured OPA server URL.
4. The OPA server evaluates the input against its loaded policies and returns a decision document.
5. PolicyAutomation inspects the result field of the decision (e.g., "allow", "warn", "fail").
6. If the result matches the trigger (e.g., result is "fail" and trigger is on_fail), it invokes the _AutomationDispatcher.
7. The _AutomationDispatcher looks up the configured actions for the result (e.g., the fail action) and executes the specified handlers via AutomationConnectors.

Integration in SSDLC Stages

Policy gating is a key part of the decision stage in the Aldeci SSDLC pipeline. The StageRunner consumes outputs from previous stages to make a final verdict.

Sources: core/stage_runner.py:530-583

Decision Stage Gating

In the _process_decision method, the system determines a final verdict of ALLOW or DEFER. A "DEFER" verdict is a form of gating, indicating the process should be paused for manual review.

The verdict is set to DEFER if:

• There are any failing controls identified in the deploy stage output.
• There are any known exploited vulnerabilities (KEV hits) in the operate stage output.

This logic acts as an internal policy gate, complementing the external OPA-based evaluation.

# core/stage_runner.py:561-565
failing_controls = [
    entry.get("control")
    for entry in deploy_manifest.get("control_evidence", [])
    if isinstance(entry, Mapping) and entry.get("result") == "fail"
]
kev_hits = operate_snapshot.get("kev_hits", []) if isinstance(operate_snapshot, Mapping) else []
verdict = "DEFER" if failing_controls or kev_hits else "ALLOW"

Sources: core/stage_runner.py:561-565

This flow is visualized below:

This diagram shows how the decision stage acts as a gate by checking for failing controls or KEV hits.
Sources: core/stage_runner.py:561-565

CI/CD Policy Evaluation

Tests for the evidence bundler also demonstrate how policies are evaluated in a CI context. A policy can be loaded and evaluated to produce a "warn" or "fail" outcome, which can be used to control pipeline execution.

# tests/test_evidence_bundle.py:50-51
def test_evaluate_policy_warn_and_fail(tmp_path):
    # ...
    # Test a 'warn' scenario
    result_warn = evaluate_policy(policy, {"risk_score": 0.6})
    assert result_warn["decision"] == "warn"

    # Test a 'fail' scenario
    result_fail = evaluate_policy(policy, {"risk_score": 0.8})
    assert result_fail["decision"] == "fail"

This test confirms that evaluate_policy returns different decisions based on the input risk_score, simulating a CI gate that checks risk thresholds.
Sources: tests/test_evidence_bundle.py:50-68