OPA / Rego: policy as code para infra e API

package authz

import future.keywords.in

# Default deny — fundamental
default allow := false

# Admin pode tudo
allow {
  "admin" in input.subject.roles
}

# Owner pode editar próprio recurso
allow {
  input.action == "update"
  input.resource.owner_id == input.subject.id
}

# Viewer pode ler dentro do horário comercial COM MFA
allow {
  "viewer" in input.subject.roles
  input.action == "read"
  input.subject.mfa == true
  business_hours
}

# Helper rule
business_hours {
  hour := time.clock(time.now_ns())[0]
  hour >= 9
  hour < 18
}

# Regra que retorna OBJETO (não só boolean)
decision := {
  "allow": allow,
  "reason": reason,
}

reason := "admin override" {
  "admin" in input.subject.roles
}
reason := "owner of resource" {
  input.resource.owner_id == input.subject.id
}
reason := "outside business hours" {
  not business_hours
}

package authz_test

import data.authz

test_admin_allowed {
  authz.allow with input as {
    "subject": { "roles": ["admin"], "id": "u1" },
    "resource": { "owner_id": "u2" },
    "action": "delete"
  }
}

test_viewer_outside_hours_denied {
  not authz.allow with input as {
    "subject": { "roles": ["viewer"], "id": "u1", "mfa": true },
    "resource": { "owner_id": "u2" },
    "action": "read"
  }
    with time.now_ns as 1700000000000000000  # 19:30 UTC — fora horário
}

test_owner_can_update {
  authz.allow with input as {
    "subject": { "roles": [], "id": "u1" },
    "resource": { "owner_id": "u1" },
    "action": "update"
  }
}

# CLI nativo
opa test -v ./policies/
# →
# PASS: 3/3 tests
#  authz_test.test_admin_allowed (1.2ms)
#  authz_test.test_viewer_outside_hours_denied (0.9ms)
#  authz_test.test_owner_can_update (0.8ms)

# Coverage
opa test --coverage --format=json ./policies/ | jq '.coverage'

services:
  bundles:
    url: https://policies.corp.example/bundles
    credentials:
      bearer:
        token: ${OPA_BUNDLE_TOKEN}

bundles:
  authz:
    service: bundles
    resource: authz.tar.gz
    polling:
      min_delay_seconds: 30
      max_delay_seconds: 60
    signing:
      keyid: prod-bundle-key

decision_logs:
  console: true
  reporting:
    min_delay_seconds: 5
    max_delay_seconds: 10

apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequiredlabels
spec:
  crd:
    spec:
      names: { kind: K8sRequiredLabels }
      validation:
        openAPIV3Schema:
          properties:
            labels: { type: array, items: { type: string } }
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredlabels

        violation[{"msg": msg, "details": {"missing_labels": missing}}] {
          provided := {label | input.review.object.metadata.labels[label]}
          required := {label | label := input.parameters.labels[_]}
          missing := required - provided
          count(missing) > 0
          msg := sprintf("missing required labels: %v", [missing])
        }
---
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sRequiredLabels
metadata: { text: ns-must-have-owner }
spec:
  match:
    kinds: [{ apiGroups: [""], kinds: ["Namespace"] }]
  parameters:
    labels: ["owner", "cost-center"]

# Policy para Envoy ext_authz
package envoy.authz

default allow := false

# Token JWT parsed do header
claims := payload {
  [_, payload, _] := io.jwt.decode(input.attributes.request.http.headers.authorization)
}

# Allow se token válido e path autorizado
allow {
  claims.sub
  some path
  path := input.attributes.request.http.path
  glob.match(allowed_paths[_], ["/"], path)
}

allowed_paths := ["/api/public/*", "/api/users/me", "/healthz"]

# Policy: "user lê docs do mesmo departamento"
# Partial eval contra subject=alice (dept=eng), sem definir resource
opa eval --partial --data policy.rego --input subject.json \
  'data.authz.allow'

# Saída (simplificada): "input.resource.dept == 'eng'"
# Esse expression é traduzido para SQL: WHERE resource.dept = 'eng'