if user.is_admin:
    allow()

if user.is_admin or user.id == project.owner_id or (
    user.region == ticket.region and ticket.status != "closed"
):
    ...

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::example-bucket/reports/*",
      "Condition": {
        "StringEquals": {
          "aws:PrincipalTag/team": "finance"
        }
      }
    }
  ]
}

subject + action + resource + context -> decision

User / Workload
  |
  v
Identity Provider / SPIRE
  |  OIDC login or SVID
  v
STS Service  <----> OpenFGA: can_assume role?
  |           <----> OPA: trust policy conditions?
  | issues short-lived role session
  v
Client with STS token
  |
  v
API Gateway / Middleware (PEP)
  |----> OpenFGA: role/session relation to resource?
  |----> OPA: permission policy, condition, explicit deny?
  |----> Audit: who/action/resource/decision/reason
  v
Business Service

workload -> SPIRE Workload API -> JWT-SVID
workload -> STS: AssumeRoleWithSVID(JWT-SVID, role)
STS -> verify SVID trust domain and SPIFFE ID
STS -> OpenFGA + OPA
STS -> short-lived role session token

spiffe://example.org/ns/billing/sa/report-generator

{
  "principal": "spiffe://example.org/ns/billing/sa/report-generator",
  "role": "role:billing-report-reader",
  "context": {
    "cluster": "prod-us-west",
    "mfa": false
  }
}

workload:spiffe://example.org/ns/billing/sa/report-generator trusted role:billing-report-reader

api-gateway --mTLS--> opa
api-gateway --mTLS--> openfga
api-gateway --mTLS--> sts

{
  "workload": {
    "spiffe_id": "spiffe://example.org/ns/agent/sa/coding-agent",
    "trust_domain": "example.org"
  },
  "action": "tool.execute",
  "resource": "tool:shell"
}

deny contains "coding agent cannot execute shell outside sandbox" if {
  input.workload.spiffe_id == "spiffe://example.org/ns/agent/sa/coding-agent"
  input.action == "tool.execute"
  input.resource == "tool:shell"
  input.context.sandbox != true
}

type user

type role
  relations
    define trusted: [user]
    define can_assume: trusted

user:alice trusted role:prod-admin

POST /assume-role
{
  "user": "user:alice",
  "role": "role:prod-admin"
}

Can user:alice can_assume role:prod-admin?

package iam.trust

default allow := false

allow if {
  input.relationship.allowed == true
  input.role == "role:prod-admin"
  input.context.mfa == true
  input.context.source == "vpn"
  input.context.ticket != ""
}

deny contains "assume prod-admin requires MFA" if {
  input.role == "role:prod-admin"
  input.context.mfa != true
}

OpenFGA: 是否存在信任关系
OPA: 当前上下文是否允许 assume
STS: 通过后签发短期 token

type user

type role
  relations
    define assignee: [user]

type secret
  relations
    define reader: [user, role#assignee]
    define deleter: [role#assignee]
    define can_read: reader
    define can_delete: deleter

user:alice assignee role:prod-admin
role:prod-admin#assignee deleter secret:prod-123_db-password

package iam.permission

default allow := false

allow if {
  input.relationship.allowed == true
  input.action == "secret.delete"
  input.resource.env == "prod"
  input.session.role == "role:prod-admin"
  input.context.mfa == true
  input.context.ticket != ""
}

deny contains "contractor cannot access confidential secrets" if {
  input.subject.type == "contractor"
  input.resource.classification == "confidential"
}

relationship says: 有资格
policy says: 此时此地可以做
explicit deny says: 就算有资格也不行

1. Alice 通过 Keycloak 登录，拿到 user token。
2. Alice 调 STS：AssumeRole(role:prod-admin)。
3. STS 校验 user token。
4. STS 调 OpenFGA：user:alice can_assume role:prod-admin?
5. STS 调 OPA：MFA、VPN、ticket、risk 是否满足 trust policy?
6. 通过后，STS 签发 15 分钟短期 token。
7. 后续请求用这个 token 访问业务 API。

{
  "sub": "user:alice",
  "role": "role:prod-admin",
  "session_id": "sess_01HV...",
  "scope": ["secret.read", "secret.delete"],
  "iat": 1714380000,
  "exp": 1714380900
}

input + data + policy -> decision

DELETE /projects/prod-123/secrets/db-password

{
  "user": {
    "id": "alice",
    "role": "developer",
    "groups": ["team-a"]
  },
  "action": "secret.delete",
  "resource": {
    "type": "secret",
    "project": "prod-123",
    "env": "prod",
    "owner_group": "team-a"
  },
  "request": {
    "mfa": true,
    "source_ip": "10.0.1.23"
  }
}

package authz

default allow := false

allow if {
  input.user.role == "admin"
  input.request.mfa == true
}

allow if {
  input.action == "secret.read"
  input.resource.owner_group in input.user.groups
}

deny_reason contains "developers cannot delete production secrets" if {
  input.action == "secret.delete"
  input.resource.env == "prod"
  input.user.role == "developer"
}

user, relation, object

user:alice reader document:roadmap
user:bob member group:security
group:security#member viewer folder:prod-secrets
folder:prod-secrets parent organization:zoom

Can user:alice read document:roadmap?

model
  schema 1.1

type user

type group
  relations
    define member: [user]

type document
  relations
    define owner: [user]
    define viewer: [user, group#member]
    define editor: [user, group#member] or owner
    define can_read: viewer or editor or owner
    define can_write: editor or owner

{
  "user": "user:alice",
  "relation": "owner",
  "object": "document:roadmap"
}

{
  "user": "group:security#member",
  "relation": "viewer",
  "object": "document:incident-runbook"
}

{
  "user": "user:alice",
  "relation": "can_read",
  "object": "document:roadmap"
}

{
  "allowed": true
}

OPA：判断请求是否符合上下文策略
OpenFGA：判断主体是否和对象存在授权关系
应用 / 网关：负责执行拦截，也就是 PEP

Client
  |
  v
API Gateway / Middleware  <-- PEP
  |
  +--> OPA：检查上下文策略
  |
  +--> OpenFGA：检查对象关系
  |
  v
Business Service

{
  "subject": "user:alice",
  "action": "secret.delete",
  "object": "secret:prod-db-password",
  "context": {
    "env": "prod",
    "mfa": true,
    "ticket": "SEC-12345",
    "data_classification": "confidential",
    "source": "vpn"
  }
}

Can user:alice delete secret:prod-db-password?

这个 delete 操作在 prod + confidential + 当前上下文下是否允许？

allow = openfga.allowed && opa.allow && not opa.deny

if user.is_admin:
    ...

HTTP Request -> FastAPI PEP -> OpenFGA Check -> OPA Decision -> allow / deny

package http.authz

default allow := false

allow if {
  input.action == "secret.read"
  input.relationship.allowed == true
}

allow if {
  input.action == "secret.delete"
  input.relationship.allowed == true
  input.resource.env == "prod"
  input.user.role == "admin"
  input.request.mfa == true
  input.request.ticket != ""
}

deny contains "OpenFGA relationship check failed" if {
  input.relationship.allowed != true
}

deny contains "read requires can_read relationship" if {
  input.action == "secret.read"
  input.relationship.allowed != true
}

deny contains "production delete requires admin" if {
  input.action == "secret.delete"
  input.resource.env == "prod"
  input.user.role != "admin"
}

deny contains "production delete requires MFA" if {
  input.action == "secret.delete"
  input.resource.env == "prod"
  input.request.mfa != true
}

deny contains "production delete requires approved ticket" if {
  input.action == "secret.delete"
  input.resource.env == "prod"
  input.request.ticket == ""
}

decision := {
  "allow": allow,
  "deny": deny,
}

fastapi
uvicorn
httpx

import os
import time
from typing import Any

import httpx
from fastapi import FastAPI, HTTPException, Request


OPA_URL = os.getenv(
    "OPA_URL",
    "http://opa:8181/v1/data/http/authz/decision",
)
OPENFGA_URL = os.getenv("OPENFGA_URL", "http://openfga:8080")

app = FastAPI()

FGA_STORE_ID = ""
FGA_MODEL_ID = ""


FGA_MODEL = {
    "schema_version": "1.1",
    "type_definitions": [
        {"type": "user"},
        {
            "type": "secret",
            "relations": {
                "reader": {"this": {}},
                "deleter": {"this": {}},
                "can_read": {
                    "union": {
                        "child": [
                            {"computedUserset": {"relation": "reader"}},
                            {"computedUserset": {"relation": "deleter"}},
                        ]
                    }
                },
                "can_delete": {"computedUserset": {"relation": "deleter"}},
            },
            "metadata": {
                "relations": {
                    "reader": {
                        "directly_related_user_types": [{"type": "user"}]
                    },
                    "deleter": {
                        "directly_related_user_types": [{"type": "user"}]
                    },
                }
            },
        },
    ],
}


FGA_TUPLES = [
    {
        "user": "user:alice",
        "relation": "reader",
        "object": "secret:prod-123_db-password",
    },
    {
        "user": "user:admin",
        "relation": "deleter",
        "object": "secret:prod-123_db-password",
    },
]


def secret_object(project: str, secret_name: str) -> str:
    return f"secret:{project}_{secret_name}"


def parse_bool(value: str | None) -> bool:
    return str(value).lower() in {"1", "true", "yes", "y"}


async def wait_for_openfga() -> None:
    for _ in range(30):
        try:
            async with httpx.AsyncClient(timeout=2.0) as client:
                response = await client.post(
                    f"{OPENFGA_URL}/stores",
                    json={"name": "authz-demo"},
                )
                response.raise_for_status()
                global FGA_STORE_ID
                FGA_STORE_ID = response.json()["id"]
                return
        except Exception:
            time.sleep(1)
    raise RuntimeError("OpenFGA is not ready")


async def init_openfga() -> None:
    await wait_for_openfga()

    async with httpx.AsyncClient(timeout=5.0) as client:
        model_response = await client.post(
            f"{OPENFGA_URL}/stores/{FGA_STORE_ID}/authorization-models",
            json=FGA_MODEL,
        )
        model_response.raise_for_status()

        global FGA_MODEL_ID
        FGA_MODEL_ID = model_response.json()["authorization_model_id"]

        tuple_response = await client.post(
            f"{OPENFGA_URL}/stores/{FGA_STORE_ID}/write",
            json={
                "authorization_model_id": FGA_MODEL_ID,
                "writes": {"tuple_keys": FGA_TUPLES},
            },
        )
        tuple_response.raise_for_status()


@app.on_event("startup")
async def startup() -> None:
    await init_openfga()


def build_authz_input(
    request: Request,
    project: str,
    secret_name: str,
    action: str,
) -> dict[str, Any]:
    return {
        "user": {
            "id": request.headers.get("x-user", "anonymous"),
            "role": request.headers.get("x-role", "developer"),
        },
        "action": action,
        "resource": {
            "type": "secret",
            "project": project,
            "name": secret_name,
            "env": "prod" if project.startswith("prod") else "dev",
            "object": secret_object(project, secret_name),
        },
        "request": {
            "mfa": parse_bool(request.headers.get("x-mfa")),
            "ticket": request.headers.get("x-ticket", ""),
            "source_ip": request.client.host if request.client else "",
        },
    }


async def check_openfga(input_doc: dict[str, Any]) -> dict[str, Any]:
    relation = "can_read"
    if input_doc["action"] == "secret.delete":
        relation = "can_delete"

    body = {
        "authorization_model_id": FGA_MODEL_ID,
        "tuple_key": {
            "user": f"user:{input_doc['user']['id']}",
            "relation": relation,
            "object": input_doc["resource"]["object"],
        },
    }

    async with httpx.AsyncClient(timeout=2.0) as client:
        response = await client.post(
            f"{OPENFGA_URL}/stores/{FGA_STORE_ID}/check",
            json=body,
        )
        response.raise_for_status()

    return {
        "relation": relation,
        "allowed": response.json().get("allowed", False),
    }


async def authorize(input_doc: dict[str, Any]) -> dict[str, Any]:
    input_doc["relationship"] = await check_openfga(input_doc)

    async with httpx.AsyncClient(timeout=2.0) as client:
        response = await client.post(OPA_URL, json={"input": input_doc})
        response.raise_for_status()

    decision = response.json().get("result", {})
    if not decision.get("allow", False):
        raise HTTPException(
            status_code=403,
            detail={
                "message": "forbidden",
                "decision": decision,
                "input": input_doc,
            },
        )

    return decision


@app.get("/projects/{project}/secrets/{secret_name}")
async def read_secret(project: str, secret_name: str, request: Request):
    input_doc = build_authz_input(request, project, secret_name, "secret.read")
    decision = await authorize(input_doc)
    return {
        "secret": f"{project}/{secret_name}",
        "value": "__REDACTED__",
        "decision": decision,
    }


@app.delete("/projects/{project}/secrets/{secret_name}")
async def delete_secret(project: str, secret_name: str, request: Request):
    input_doc = build_authz_input(request, project, secret_name, "secret.delete")
    decision = await authorize(input_doc)
    return {
        "deleted": f"{project}/{secret_name}",
        "decision": decision,
    }

FROM python:3.12-slim

WORKDIR /app

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

COPY app.py .

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

services:
  opa:
    image: openpolicyagent/opa:latest
    command:
      - run
      - --server
      - --addr
      - 0.0.0.0:8181
      - /policy.rego
    volumes:
      - ./policy.rego:/policy.rego:ro
    ports:
      - "8181:8181"

  openfga:
    image: openfga/openfga:latest
    command: ["run"]
    ports:
      - "8080:8080"
      - "8081:8081"
      - "3000:3000"

  api:
    build: .
    environment:
      OPA_URL: http://opa:8181/v1/data/http/authz/decision
      OPENFGA_URL: http://openfga:8080
    depends_on:
      - opa
      - openfga
    ports:
      - "8000:8000"

docker compose up --build

curl -i \
  -H 'X-User: alice' \
  -H 'X-Role: developer' \
  http://localhost:8000/projects/prod-123/secrets/db-password

curl -i \
  -H 'X-User: bob' \
  -H 'X-Role: developer' \
  http://localhost:8000/projects/prod-123/secrets/db-password

{
  "message": "forbidden",
  "decision": {
    "allow": false,
    "deny": [
      "OpenFGA relationship check failed",
      "read requires can_read relationship"
    ]
  }
}

curl -i -X DELETE \
  -H 'X-User: alice' \
  -H 'X-Role: developer' \
  -H 'X-MFA: true' \
  -H 'X-Ticket: SEC-12345' \
  http://localhost:8000/projects/prod-123/secrets/db-password

curl -i -X DELETE \
  -H 'X-User: admin' \
  -H 'X-Role: admin' \
  -H 'X-MFA: true' \
  -H 'X-Ticket: SEC-12345' \
  http://localhost:8000/projects/prod-123/secrets/db-password

subject + action + resource + context

FastAPI PEP
  -> OpenFGA check(user, can_read/can_delete, secret)
  -> OPA eval(subject/action/resource/context + relationship.allowed)
  -> allow / deny

JWT valid -> allow

Authentication -> Identity
Authorization -> Decision

{ "allowed": false }

{
  "allowed": false,
  "reason": "production delete requires approved ticket"
}

所有高危 API：
1. 必须经过统一 PEP；
2. 必须有 subject/workload/session/role/action/resource/context；
3. 服务间调用必须有可验证 workload identity，例如 SPIFFE ID；
4. 必须检查 OpenFGA relationship；
5. 必须检查 OPA condition 和 explicit deny；
6. 必须记录 allow/deny 和 reason；
7. deny 默认安全，不允许策略服务失败时放行；
8. 策略变更必须可 review、可回滚。

@startmindmap
* OSS IAM-like 授权系统
** AWS IAM 心智模型
*** Principal
*** Role / AssumeRole
*** Trust Policy
*** Permission Policy
*** Resource Policy
*** Condition / Explicit Deny
*** CloudTrail
** 开源组件
*** Keycloak / Dex / ORY
**** 身份登录和联邦
*** SPIFFE / SPIRE
**** workload identity
**** X.509-SVID / JWT-SVID
**** mTLS
*** 自建 STS
**** AssumeRole
**** AssumeRoleWithSVID
**** 短期 session token
*** OpenFGA
**** can_assume trust relationship
**** resource relationship
**** Check / Batch Check
*** OPA
**** permission policy
**** condition
**** explicit deny
**** guardrail
*** PEP
**** API Gateway
**** middleware
*** Audit
**** decision log
**** reason / policy version
** 授权流程
*** Login
*** Workload gets SVID
*** AssumeRole
*** STS verifies SPIFFE ID
*** STS check OpenFGA
*** STS eval OPA trust policy
*** Issue short-lived token
*** API PEP checks OpenFGA + OPA
** 可运行 HTTP 示例
*** docker-compose 启动 api / opa / openfga
*** FastAPI 作为 PEP
*** OpenFGA 初始化关系模型
*** OPA REST API 决策
*** curl 验证 200 / 403
** 落地要点
*** 先接身份源
*** 接 SPIFFE/SPIRE
*** 设计 role/resource 模型
*** 最小 STS
*** 策略版本化
*** 单元测试
*** 审计与解释
*** 性能与缓存
** 常见坑
*** 认证当授权
*** 长期 token 当 session
*** OpenFGA 承担所有条件策略
*** OPA 变成关系数据库
*** 策略服务失败时默认放行
@endmindmap