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

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

```rego
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

```rego test
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"}),
  })
}
```