Configure Authorization Policies

apps:
  - name: my-oidc-app
    type: oidc
    authorization:
      tokenMinting:
        accessToken:
          policies:
            - name: token-governance
              rego: |
                package orchestrator

                default result := {"allowed": true}

                # Deny tokens requested with the client_credentials grant type
                result := {"allowed": false, "internal_message": "client_credentials not permitted"} {
                  input.request.oauth.grant_type == "client_credentials"
                }

package orchestrator

default result["allowed"] := false

# Allow all non-token-exchange grant types.
result["allowed"] if {
	input.request.oauth.grant_type != "urn:ietf:params:oauth:grant-type:token-exchange"
}

# Parse scopes from space-separated strings into sets.
requested_scopes := {s | some s in split(input.request.oauth.scope, " "); s != ""}
subject_scopes := {s | some s in split(input.request.oauth.subject_token.claims.scope, " "); s != ""}

# For token exchange grants, allow if requested scopes are a subset of subject token scopes.
result["allowed"] if {
	input.request.oauth.grant_type == "urn:ietf:params:oauth:grant-type:token-exchange"
	print("requested_scopes:", requested_scopes)
	print("subject_scopes:", subject_scopes)
	count(requested_scopes - subject_scopes) == 0
}

result["internal_message"] := "requested scopes exceed subject_token scopes" if {
	not result.allowed
}

result["external_message"] := "invalid_scope" if {
	not result.allowed
}

package orchestrator

# Default deny policy - all requests are denied unless explicitly allowed
default result["allowed"] := false

# Decode the bearer token's claims. jwt.parse_bearer strips the "Bearer "
# prefix and returns the payload, or undefined if the header is absent or
# does not carry a JWT -- in which case every rule below fails and the
# default deny applies.
jwt_payload := jwt.parse_bearer(input.request.http.headers.Authorization)

# Allows access to the get_ticket_price tool if the token contains
# the tickets:read scope. Logs the tool name, client ID, and subject
# for audit purposes.
result["allowed"] if {
	print("request made to tool: ", input.request.mcp.tool.params.name)
	input.request.mcp.tool.params.name == "get_ticket_price"

	print("request made with client of: ", jwt_payload.client_id)
	contains(jwt_payload.scope, "tickets:read")

	print("access granted to subject:", jwt_payload.sub)
}

result["internal_message"] := "access is only permitted to the 'get_ticket_price' tool with tickets:read scope" if {
	not result.allowed
}

result["external_message"] := "unauthorized, contact support@example.com" if {
	not result.allowed
}

package orchestrator

default result["allowed"] := false

jwt_payload := jwt.parse_bearer(input.request.http.headers.Authorization)

result["allowed"] if {
	"mission-control" in jwt_payload.permitted_mcp_servers
}

result["internal_message"] := "bearer token does not permit access to this MCP server" if {
	not result["allowed"]
}

result["external_message"] := "unauthorized: MCP server access not granted" if {
	not result["allowed"]
}

package orchestrator

default result["allowed"] := false

# Tools visible to downstream MCP clients
allowed_tools := {"listUsers", "getUser", "searchUsers"}

result["allowed"] if {
	input.request.mcp.type == "tools_list"
	input.request.mcp.response.tools_list.name in allowed_tools
}

result["internal_message"] := "tool is not in the allowed set" if {
	not result["allowed"]
}

package orchestrator

default result["allowed"] := false

jwt_payload := jwt.parse_bearer(input.request.http.headers.Authorization)

# Read-only tools available to agents with ticketService:read or ticketService:write
read_tools := {"listTickets", "getTicket", "searchTickets"}

# Write tools restricted to agents with the ticketService:write scope
write_tools := {"createTicket", "updateTicket", "closeTicket"}

result["allowed"] if {
	input.request.mcp.type == "tools_list"
	input.request.mcp.response.tools_list.name in read_tools
	some s in ["ticketService:read", "ticketService:write"]
	contains(jwt_payload.scope, s)
}

result["allowed"] if {
	input.request.mcp.type == "tools_list"
	input.request.mcp.response.tools_list.name in write_tools
	contains(jwt_payload.scope, "ticketService:write")
}

result["internal_message"] := "tool not permitted for this agent's scope" if {
	not result["allowed"]
}