access-engine/examples/ops-warden-fetch/policy_package.md

id	name	namespace	version	status	package	allow_ttl	actions	owner	fixtures	caring	activation	metadata
ops-warden.secret.read	Ops-Warden secret read (catalog fetch lanes)	ops-warden:secret-read	v1	ready	flexauth.ops_warden.secret_read	5m	read	team:platform-security	policy_fixtures.yaml	profile enforce canonical_roles organization_relations scopes planes capabilities exposure_modes conditions restrictions caring-0.4.0-rc2 false Operator ServiceProvider level id tenant Platform platform:secret-read tenant:platform Secret Audit View Use SecretMaterial PurposeBound TimeLimited Logged PrivilegeEscalationBlocked	mode local	source generator catalog origin examples/ops-warden-fetch/policy_package.md tools/gen_ops_warden_fetch.py ops-warden/registry/routing/catalog.yaml CUST-WP-0075 (bug b792c5d0)

Ops-Warden Secret Read

ops-warden's warden access <need> --fetch/--exec asks this package, before it proxies a non-SSH credential read, whether read on secret:<need_id> owned by <owner_repo> is allowed (ops-warden src/warden/policy.py check_fetch_policy). ops-warden is the conduit, never the owner; the owner system is the lane's owner_repo, and no secret value is in the request.

The readable set is not authored here. catalog_lanes maps each need id to its owner system and is generated from ops-warden's routing catalog: every entry with status: active, exec_capable: true and lane: secret. Regenerate with python3 tools/gen_ops_warden_fetch.py; --check reports drift.

The set lives in the package rather than being read from registry attributes so that a caller cannot assert its way onto a lane the registry does not know (the engine passes caller attributes through for unregistered resources, FLEX-DEC-2026-012).

Who may ask is enforced before this package runs: the pin's caller bindings map every owner system to the ops-warden ServiceAccount, in enforce mode, so any other caller gets HTTP 403 and an owner system absent from the bindings gets 403 as well. The package repeats the system check as unknown_system / wrong_system for offline evaluation and defence in depth.

Rules

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

# BEGIN GENERATED catalog_lanes (tools/gen_ops_warden_fetch.py; do not edit by hand)
catalog_lanes := {
  "agent-harness-forgejo-deploy": "railiance-platform",
  "binky-company-email-imap": "railiance-platform",
  "binky-qonto-api": "railiance-platform",
  "email-connect-transactional": "railiance-platform",
  "forgejo-admin-api-token": "railiance-platform",
  "issue-core-ingestion-api-key": "railiance-platform",
  "openrouter-llm-connect": "railiance-platform",
  "policy-nexus-forgejo-source-read": "railiance-platform",
  "railiance-backup-offsite-lane": "railiance-platform",
  "rapp-qonto-keycape-client": "key-cape",
  "reuse-surface-hub-write-token": "railiance-platform",
  "whynot-design-npm-publish": "railiance-platform",
}
# END GENERATED catalog_lanes

catalog_systems := {system | system := catalog_lanes[_]}

decision := {"effect": "allow", "reason": "secret_read_lane_matched"} if {
  allowed
} else := {"effect": "deny", "reason": first_denial} if {
  true
}

allowed if {
  input.action == "read"
  input.resource.type == "secret"
  effective_tenant == "tenant:platform"
  input.subject.type == "operator"
  has_subject_id
  need_known
  catalog_lanes[need_id] == input.resource.system
  resource_id_matches_need
  context_owner_matches
}

default need_id := ""

need_id := input.context.need_id if {
  is_string(input.context.need_id)
}

need_known if {
  need_id != ""
  catalog_lanes[need_id]
}

has_subject_id if {
  is_string(input.subject.id)
  input.subject.id != ""
}

default effective_tenant := ""

effective_tenant := input.tenant if {
  is_string(input.tenant)
  input.tenant != ""
} else := input.resource.tenant if {
  is_string(input.resource.tenant)
  input.resource.tenant != ""
} else := input.subject.tenant if {
  is_string(input.subject.tenant)
  input.subject.tenant != ""
}

# warden sends secret:<need_id>, or secret:<need_id>/<domain> when a domain is set.
resource_id_matches_need if {
  not has_domain
  input.resource.id == sprintf("secret:%s", [need_id])
}

resource_id_matches_need if {
  has_domain
  input.resource.id == sprintf("secret:%s/%s", [need_id, input.context.domain])
}

has_domain if {
  is_string(input.context.domain)
  input.context.domain != ""
}

context_owner_matches if {
  input.context.owner_repo == input.resource.system
}

default first_denial := "no_matching_rule"

first_denial := "wrong_action" if {
  input.action != "read"
} else := "wrong_resource_type" if {
  input.resource.type != "secret"
} else := "wrong_tenant" if {
  effective_tenant != "tenant:platform"
} else := "wrong_subject_type" if {
  input.subject.type != "operator"
} else := "missing_subject" if {
  not has_subject_id
} else := "missing_need_id" if {
  need_id == ""
} else := "unknown_need" if {
  not need_known
} else := "unknown_system" if {
  not input.resource.system in catalog_systems
} else := "wrong_system" if {
  catalog_lanes[need_id] != input.resource.system
} else := "resource_need_mismatch" if {
  not resource_id_matches_need
} else := "owner_repo_mismatch" if {
  not context_owner_matches
}

Tests

package flexauth.ops_warden.secret_read_test

import future.keywords.if
import data.flexauth.ops_warden.secret_read

forgejo := {
  "subject": {"id": "operator", "type": "operator", "tenant": "tenant:platform"},
  "action": "read",
  "resource": {
    "id": "secret:forgejo-admin-api-token",
    "type": "secret",
    "system": "railiance-platform",
    "tenant": "tenant:platform"
  },
  "context": {"need_id": "forgejo-admin-api-token", "owner_repo": "railiance-platform", "domain": null}
}

test_forgejo_admin_read_allowed if {
  secret_read.decision.effect == "allow" with input as forgejo
}

test_domain_suffix_allowed if {
  secret_read.decision.effect == "allow" with input as object.union(forgejo, {
    "resource": object.union(forgejo.resource, {"id": "secret:forgejo-admin-api-token/infotech"}),
    "context": object.union(forgejo.context, {"domain": "infotech"}),
  })
}

test_unknown_need_denied if {
  secret_read.decision.reason == "unknown_need" with input as object.union(forgejo, {
    "resource": object.union(forgejo.resource, {"id": "secret:no-such-need"}),
    "context": object.union(forgejo.context, {"need_id": "no-such-need"}),
  })
}

test_unknown_system_denied if {
  secret_read.decision.reason == "unknown_system" with input as object.union(forgejo, {
    "resource": object.union(forgejo.resource, {"system": "ops-warden"}),
    "context": object.union(forgejo.context, {"owner_repo": "ops-warden"}),
  })
}

test_wrong_action_denied if {
  secret_read.decision.reason == "wrong_action" with input as object.union(forgejo, {"action": "write"})
}

test_spoofed_domain_denied if {
  secret_read.decision.reason == "resource_need_mismatch" with input as object.union(forgejo, {
    "resource": object.union(forgejo.resource, {"id": "secret:forgejo-admin-api-token/other"}),
    "context": object.union(forgejo.context, {"domain": "infotech"}),
  })
}