Skip to main content
← Back to Blog

Building Policy-Driven Applications: Using Sidecar with OPA in Python

A Comprehensive Guide to Implementing Authorization as a Service

Introduction

As modern applications grow in complexity, managing authorization and access control becomes increasingly challenging. The sidecar pattern combined with Open Policy Agent (OPA) offers an elegant solution by decoupling policy decisions from application code. This architectural approach enables you to centralize policy management, improve security, and maintain flexibility as your authorization requirements evolve.

In this blog post, we'll explore how to implement a Python application that leverages OPA as a sidecar for policy enforcement. We'll cover the architecture, implementation details, and best practices for building production-ready, policy-driven applications.

What is Open Policy Agent (OPA)?

Open Policy Agent is a general-purpose policy engine that enables unified, context-aware policy enforcement across your entire stack. OPA provides a high-level declarative language called Rego for writing policies and REST APIs for querying those policies.

Key benefits of OPA include:

  • Decoupling policy decisions from application code
  • Policy reusability across multiple services
  • Centralized policy management and versioning
  • Support for complex authorization scenarios
  • Language-agnostic policy enforcement

Understanding the Sidecar Pattern

The sidecar pattern is a container design pattern where a helper container (the sidecar) is deployed alongside your main application container. In the context of OPA, the sidecar runs as a separate process that handles all policy evaluation requests from your application.

This pattern provides several advantages:

  • Separation of concerns: Policy logic is isolated from business logic
  • Independent scaling: OPA can be scaled separately from your application
  • Technology independence: OPA works with any programming language
  • Simplified updates: Policies can be updated without redeploying applications

Architecture Overview

Our implementation consists of three main components:

  1. Python Application: The main service that needs authorization decisions
  2. OPA Sidecar: A containerized OPA instance running alongside the application
  3. Rego Policies: Declarative policy definitions that encode authorization rules

The application communicates with OPA through HTTP REST API calls, sending input data and receiving policy decisions in JSON format. This loose coupling ensures that policy changes don't require application code modifications.

Implementation Guide

Step 1: Setting Up the OPA Sidecar

First, we'll set up OPA as a Docker container. Create a docker-compose.yml file:

services:
  opa:
    image: openpolicyagent/opa:latest
    ports:
      - "8181:8181"
    volumes:
      - ./policies:/policies
    command:
      - "run"
      - "--server"
      - "--addr=0.0.0.0:8181"
      - "--log-level=debug"
      - "/policies"

  app:
    build: .
    ports:
      - "5001:5001"
    environment:
      - OPA_URL=http://opa:8181
    depends_on:
      - opa

Step 2: Writing Rego Policies

Create a policy file (policies/authz.rego) that defines authorization rules. Here's an example for role-based access control:

package authz

import rego.v1

# Default deny
default allow := false

# Allow admins to perform any action
allow if {
    input.user.role == "admin"
}

# Allow users to read their own resources
allow if {
    input.action == "read"
    input.resource.owner == input.user.id
}

# Allow users to update their own resources
allow if {
    input.action == "update"
    input.resource.owner == input.user.id
    input.user.role == "user"
}

# Allow managers to read resources in their department
allow if {
    input.action == "read"
    input.user.role == "manager"
    input.resource.department == input.user.department
}

Step 3: Creating the Python Client

Now, let's create a Python client to interact with OPA. We'll build a reusable class that handles policy queries:

# opa_client.py
import requests
from typing import Dict, Any, Optional
import logging


class OPAClient:
    """Client for interacting with Open Policy Agent."""

    def __init__(self, opa_url: str = "http://127.0.0.1:8181", timeout: int = 5):
        """
        Initialize the OPA client.

        Args:
            opa_url: Base URL for the OPA service
            timeout: Request timeout in seconds
        """
        self.opa_url = opa_url.rstrip('/')
        self.timeout = timeout
        self.logger = logging.getLogger(__name__)

    def check_permission(
        self,
        policy_path: str,
        input_data: Dict[str, Any]
    ) -> bool:
        """
        Check if a permission is allowed based on OPA policy.

        Args:
            policy_path: Path to the policy (e.g., "authz/allow")
            input_data: Input data for policy evaluation

        Returns:
            True if allowed, False otherwise
        """
        try:
            url = f"{self.opa_url}/v1/data/{policy_path}"

            self.logger.debug(f"Querying OPA at {url} with input: {input_data}")

            response = requests.post(
                url,
                json={"input": input_data},
                timeout=self.timeout
            )
            response.raise_for_status()

            result = response.json()
            allowed = result.get("result", False)

            self.logger.info(
                f"Policy decision for {policy_path}: {'ALLOW' if allowed else 'DENY'}"
            )

            return allowed

        except requests.exceptions.Timeout:
            self.logger.error(f"OPA request timed out after {self.timeout}s")
            return False

        except requests.exceptions.RequestException as e:
            self.logger.error(f"OPA request failed: {e}")
            return False

        except Exception as e:
            self.logger.error(f"Unexpected error during policy check: {e}")
            return False

    def query_policy(
        self,
        policy_path: str,
        input_data: Dict[str, Any]
    ) -> Optional[Dict[str, Any]]:
        """
        Query OPA policy and return full result.

        Args:
            policy_path: Path to the policy
            input_data: Input data for policy evaluation

        Returns:
            Full policy result or None on error
        """
        try:
            url = f"{self.opa_url}/v1/data/{policy_path}"

            response = requests.post(
                url,
                json={"input": input_data},
                timeout=self.timeout
            )
            response.raise_for_status()

            return response.json().get("result")

        except Exception as e:
            self.logger.error(f"Policy query failed: {e}")
            return None

    def health_check(self) -> bool:
        """
        Check if OPA is healthy and reachable.

        Returns:
            True if healthy, False otherwise
        """
        try:
            url = f"{self.opa_url}/health"
            response = requests.get(url, timeout=2)
            return response.status_code == 200
        except Exception:
            return False

Step 4: Integrating with Your Application

Here's an example of how to use the OPA client in a Flask application:

# app.py
import os
from flask import Flask, request, jsonify, g
from functools import wraps
from opa_client import OPAClient
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

# Initialize OPA client
opa_url = os.environ.get("OPA_URL", "http://127.0.0.1:8181")
opa = OPAClient(opa_url=opa_url)


def get_current_user(request):
    """Extract user information from request (e.g., from JWT token)."""
    # This is a simplified example - implement proper authentication
    return {
        'id': request.headers.get('X-User-ID', 'anonymous'),
        'role': request.headers.get('X-User-Role', 'guest'),
        'department': request.headers.get('X-User-Department', 'none')
    }


def require_permission(action: str):
    """Decorator to check OPA permissions before executing endpoint."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user = get_current_user(request)

            # Extract resource information from the route
            resource_id = kwargs.get('resource_id')

            if not resource_id:
                return jsonify({"error": "Resource ID required"}), 400

            # Prepare input for OPA
            input_data = {
                "user": user,
                "action": action,
                "resource": {
                    "id": resource_id,
                    "owner": get_resource_owner(resource_id),
                    "department": get_resource_department(resource_id)
                }
            }

            # Check permission
            if not opa.check_permission("authz/allow", input_data):
                return jsonify({
                    "error": "Forbidden",
                    "message": "You don't have permission to perform this action"
                }), 403

            # Store user in g for use in the endpoint
            g.current_user = user

            return f(*args, **kwargs)

        return decorated_function
    return decorator


# Mock database functions
def get_resource_owner(resource_id: str) -> str:
    """Get the owner of a resource."""
    # In production, query your database
    mock_db = {
        "resource_1": "user_123",
        "resource_2": "user_456",
        "resource_3": "user_123"
    }
    return mock_db.get(resource_id, "unknown")


def get_resource_department(resource_id: str) -> str:
    """Get the department of a resource."""
    mock_db = {
        "resource_1": "engineering",
        "resource_2": "sales",
        "resource_3": "engineering"
    }
    return mock_db.get(resource_id, "unknown")


def get_resource_data(resource_id: str) -> dict:
    """Get resource data."""
    return {
        "id": resource_id,
        "name": f"Resource {resource_id}",
        "owner": get_resource_owner(resource_id),
        "department": get_resource_department(resource_id),
        "data": "sensitive information"
    }


# API Endpoints
@app.route('/api/resources/<resource_id>', methods=['GET'])
@require_permission('read')
def get_resource(resource_id):
    """Get a resource by ID."""
    return jsonify(get_resource_data(resource_id))


@app.route('/api/resources/<resource_id>', methods=['PUT'])
@require_permission('update')
def update_resource(resource_id):
    """Update a resource."""
    data = request.get_json()
    # Perform update logic here
    return jsonify({
        "message": "Resource updated successfully",
        "resource_id": resource_id,
        "updated_by": g.current_user['id']
    })


@app.route('/api/resources/<resource_id>', methods=['DELETE'])
@require_permission('delete')
def delete_resource(resource_id):
    """Delete a resource."""
    # Perform delete logic here
    return jsonify({
        "message": "Resource deleted successfully",
        "resource_id": resource_id
    })


@app.route('/health', methods=['GET'])
def health():
    """Health check endpoint."""
    opa_healthy = opa.health_check()

    return jsonify({
        "status": "healthy" if opa_healthy else "degraded",
        "opa": "up" if opa_healthy else "down"
    }), 200 if opa_healthy else 503


if __name__ == '__main__':
    port = int(os.environ.get("PORT", 5001))
    app.run(debug=True, host='0.0.0.0', port=port)

Step 5: Creating a Dockerfile

Create a Dockerfile for your Python application:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ .

EXPOSE 5001

CMD ["python", "app.py"]

And a requirements.txt:

flask==3.0.0
requests==2.31.0

Best Practices

1. Policy Design

  • Use a default-deny approach: Start with default allow = false to ensure security by default
  • Keep policies modular: Organize policies by domain (authz, data_filtering, etc.)
  • Write comprehensive tests: Use OPA's built-in testing framework
  • Document policy intent: Add comments explaining the business rules
  • Version your policies: Use Git to track policy changes
  • Use meaningful variable names: Make policies readable and maintainable

2. Performance Optimization

  • Implement caching: Cache policy decisions for frequently accessed resources
  • Use connection pooling: Reuse HTTP connections to OPA
  • Monitor response times: Set up alerts for slow policy evaluations
  • Consider bundle-based distribution: For large deployments, use OPA bundles
  • Optimize Rego policies: Avoid expensive operations in hot paths
  • Use partial evaluation: For data filtering scenarios, leverage OPA's partial evaluation

3. Error Handling

  • Fail closed: Always deny access when OPA is unreachable
  • Implement proper logging: Log all policy decisions for audit trails
  • Set up health checks: Monitor OPA availability continuously
  • Use circuit breakers: Prevent cascading failures when OPA is down
  • Provide meaningful error messages: Help users understand why access was denied
  • Implement retry logic: With exponential backoff for transient failures

4. Security Considerations

  • Secure communication: Use TLS for OPA API calls in production
  • Validate input data: Sanitize data before sending to OPA
  • Implement authentication: Secure OPA API access with tokens or mTLS
  • Regularly update OPA: Stay current with security patches
  • Audit policy changes: Track who changed what policies and when
  • Separate policy environments: Use different OPA instances for dev/staging/prod

5. Testing

  • Unit test policies: Write Rego tests for all policy rules
  • Integration test the client: Test OPA client error handling
  • Load test: Ensure OPA can handle your expected traffic
  • Test failure scenarios: Verify behavior when OPA is unavailable
  • Use test fixtures: Create reusable test data for policy testing

Advanced Features

Policy Bundles

For production deployments, OPA supports policy bundles that can be served from remote locations. This allows you to update policies without restarting OPA:

# Start OPA with bundle configuration
opa run --server \
  --set bundles.authz.service=bundle_service \
  --set bundles.authz.resource=bundles/authz.tar.gz \
  --set services.bundle_service.url=https://bundle-server.example.com

Create a bundle configuration file (bundle-config.json):

{
  "services": {
    "bundle_service": {
      "url": "https://bundle-server.example.com",
      "credentials": {
        "bearer": {
          "token": "your-auth-token"
        }
      }
    }
  },
  "bundles": {
    "authz": {
      "service": "bundle_service",
      "resource": "bundles/authz.tar.gz",
      "polling": {
        "min_delay_seconds": 60,
        "max_delay_seconds": 120
      }
    }
  }
}

Decision Logging

Enable decision logging to audit all policy decisions for compliance and debugging:

# Enable console decision logging
opa run --server \
  --set decision_logs.console=true

# Or send to a remote service
opa run --server \
  --set decision_logs.service=logger \
  --set services.logger.url=https://logger.example.com

Decision logs include:

  • Input data sent to the policy
  • Policy decision result
  • Timestamp and duration
  • Policy version

Policy Testing

OPA provides a built-in testing framework. Create test files alongside your policies:

# authz_test.rego
package authz

import rego.v1

# Test admin access
test_admin_allowed if {
    allow with input as {
        "user": {"role": "admin"},
        "action": "delete",
        "resource": {"owner": "someone_else"}
    }
}

# Test user can read own resources
test_user_read_own if {
    allow with input as {
        "user": {"id": "user_123", "role": "user"},
        "action": "read",
        "resource": {"owner": "user_123"}
    }
}

# Test user cannot read others' resources
test_user_cannot_read_others if {
    not allow with input as {
        "user": {"id": "user_123", "role": "user"},
        "action": "read",
        "resource": {"owner": "user_456"}
    }
}

# Test user can update own resources
test_user_update_own if {
    allow with input as {
        "user": {"id": "user_123", "role": "user"},
        "action": "update",
        "resource": {"owner": "user_123"}
    }
}

# Test user cannot update others' resources
test_user_cannot_update_others if {
    not allow with input as {
        "user": {"id": "user_123", "role": "user"},
        "action": "update",
        "resource": {"owner": "user_456"}
    }
}

# Test manager can read department resources
test_manager_department_access if {
    allow with input as {
        "user": {
            "id": "mgr_123",
            "role": "manager",
            "department": "engineering"
        },
        "action": "read",
        "resource": {
            "owner": "user_456",
            "department": "engineering"
        }
    }
}

# Test manager cannot read resources from other departments
test_manager_cannot_read_other_department if {
    not allow with input as {
        "user": {
            "id": "mgr_123",
            "role": "manager",
            "department": "engineering"
        },
        "action": "read",
        "resource": {
            "owner": "user_456",
            "department": "sales"
        }
    }
}

# Test guest cannot access anything
test_guest_denied if {
    not allow with input as {
        "user": {"id": "guest_1", "role": "guest"},
        "action": "read",
        "resource": {"owner": "user_123"}
    }
}

Run tests with:

opa test policies/ -v

Data Filtering with Partial Evaluation

OPA can help filter large datasets by generating optimized queries:

# data_filter.rego
package data_filter

import future.keywords.if
import future.keywords.in

# Return all documents the user can access
allowed_documents[doc] {
    doc := data.documents[_]
    allow_document(doc)
}

# Allow if user owns the document
allow_document(doc) if {
    doc.owner == input.user.id
}

# Allow if user is in the same department
allow_document(doc) if {
    input.user.role == "manager"
    doc.department == input.user.department
}

# Allow if user is admin
allow_document(doc) if {
    input.user.role == "admin"
}

Deployment Considerations

Kubernetes Deployment

Deploy OPA as a sidecar container in your pod definition:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
    spec:
      containers:
      - name: app
        image: myapp:latest
        ports:
        - containerPort: 5000
        env:
        - name: OPA_URL
          value: "http://localhost:8181"
        
      - name: opa
        image: openpolicyagent/opa:latest
        args:
        - "run"
        - "--server"
        - "--addr=localhost:8181"
        - "--set=decision_logs.console=true"
        ports:
        - name: http
          containerPort: 8181
        livenessProbe:
          httpGet:
            path: /health
            port: 8181
          initialDelaySeconds: 5
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health?bundles
            port: 8181
          initialDelaySeconds: 5
          periodSeconds: 10
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"

High Availability

For high-availability scenarios, consider these approaches:

Option 1: Sidecar per instance (Recommended)

  • Deploy one OPA sidecar with each application instance
  • Provides maximum isolation and performance
  • No network latency between app and OPA
  • Scales automatically with your application

Option 2: Shared OPA service

  • Run OPA as a separate deployment with multiple replicas
  • Use a Kubernetes service for load balancing
  • Reduces resource overhead
  • Requires network calls between app and OPA

Option 3: Hybrid approach

  • Use sidecars for critical paths
  • Use shared OPA for less critical operations
  • Balance performance and resource utilization

Configuration Management

Use ConfigMaps for OPA configuration:

apiVersion: v1
kind: ConfigMap
metadata:
  name: opa-config
data:
  config.yaml: |
    decision_logs:
      console: true
    bundles:
      authz:
        service: bundle_service
        resource: bundles/authz.tar.gz
    services:
      bundle_service:
        url: https://bundle-server.example.com

Reference in your deployment:

- name: opa
  image: openpolicyagent/opa:latest
  args:
  - "run"
  - "--server"
  - "--config-file=/config/config.yaml"
  volumeMounts:
  - name: opa-config
    mountPath: /config
volumes:
- name: opa-config
  configMap:
    name: opa-config

Monitoring and Observability

Prometheus Metrics

OPA exposes Prometheus metrics at the /metrics endpoint. Key metrics to monitor:

# Policy evaluation latency
http_request_duration_seconds{handler="v1/data"}

# Total number of policy decisions
http_request_count_total{handler="v1/data"}

# Bundle loading status
bundle_loaded_timestamp_seconds

# Policy compile time
rego_load_bundles_elapsed_time_seconds

Setting up Prometheus Scraping

Add Prometheus annotations to your pod:

metadata:
  annotations:
    prometheus.io/scrape: "true"
    prometheus.io/port: "8181"
    prometheus.io/path: "/metrics"

Grafana Dashboard

Create a Grafana dashboard to visualize:

  • Policy evaluation latency (p50, p95, p99)
  • Request rate and error rate
  • Bundle update status
  • OPA memory and CPU usage
  • Policy decision distribution (allow vs deny)

Alerting

Set up alerts for:

# High latency
- alert: OPAPolicyEvaluationSlow
  expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket{handler="v1/data"}[5m])) > 0.5
  annotations:
    summary: "OPA policy evaluation is slow"

# High error rate
- alert: OPAHighErrorRate
  expr: rate(http_request_count_total{handler="v1/data",code=~"5.."}[5m]) > 0.01
  annotations:
    summary: "OPA is returning errors"

# Bundle not updating
- alert: OPABundleNotUpdating
  expr: time() - bundle_loaded_timestamp_seconds > 3600
  annotations:
    summary: "OPA bundle hasn't updated in over an hour"

Distributed Tracing

Integrate OPA with your distributed tracing system:

from opentelemetry import trace
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# Instrument requests library
RequestsInstrumentor().instrument()

class OPAClient:
    def check_permission(self, policy_path: str, input_data: Dict[str, Any]) -> bool:
        tracer = trace.get_tracer(__name__)
        
        with tracer.start_as_current_span("opa.check_permission") as span:
            span.set_attribute("policy.path", policy_path)
            span.set_attribute("user.id", input_data.get("user", {}).get("id"))
            
            # Make OPA request (automatically traced)
            result = self._make_request(policy_path, input_data)
            
            span.set_attribute("policy.result", result)
            return result

Example: Complete Project Structure

Here's a recommended project structure:

opa-python-app/
├── app/
│   ├── __init__.py
│   ├── app.py                  # Flask application
│   ├── opa_client.py           # OPA client
│   ├── models.py               # Data models
│   └── decorators.py           # Auth decorators
├── policies/
│   ├── authz.rego              # Authorization policies
│   ├── authz_test.rego         # Policy tests
│   └── data_filter.rego        # Data filtering policies
├── tests/
│   ├── test_app.py             # Application tests
│   ├── test_opa_client.py      # OPA client tests
│   └── integration/
│       └── test_authorization.py
├── k8s/
│   ├── deployment.yaml         # Kubernetes deployment
│   ├── service.yaml            # Kubernetes service
│   └── configmap.yaml          # OPA configuration
├── docker-compose.yml          # Local development
├── Dockerfile                  # Application container
├── requirements.txt            # Python dependencies
└── README.md                   # Documentation

Working Example Repository: A complete, working implementation of this tutorial is available at github.com/mshoaibiqbal/opa-python-sidecar. Clone it to get started quickly with OPA sidecars in Python.

Conclusion

Using OPA as a sidecar with Python applications provides a powerful, flexible approach to authorization. By decoupling policy from code, you gain:

  • Agility: Update authorization rules without code changes
  • Consistency: Enforce the same policies across all services
  • Auditability: Clear trail of all policy decisions
  • Scalability: Independent scaling of policy evaluation
  • Security: Centralized, expert-reviewed policies

The sidecar pattern, combined with OPA's expressive policy language and robust ecosystem, enables you to build secure, compliant applications that evolve with your business requirements. As you scale this architecture, remember to focus on:

  • Comprehensive policy testing
  • Performance monitoring and optimization
  • Security hardening (TLS, authentication, input validation)
  • High availability and disaster recovery
  • Clear documentation and team training

Whether you're building microservices, APIs, or monolithic applications, OPA sidecars offer a production-ready solution for policy enforcement that integrates seamlessly with Python and any other technology in your stack.

Additional Resources

This blog post provides a foundation for implementing OPA sidecars with Python. For production deployments, always consult the official OPA documentation and consider your specific security and compliance requirements.