필사 모드: 认证、授权，以及 IDOR — 登录明明是正常的，却能看到别人的数据

04:11:52  GET /v1/orders/10421  200  user=usr_8f21  tenant=tnt_a91  12ms
04:11:58  GET /v1/orders/10422  200  user=usr_8f21  tenant=tnt_a91  9ms
04:12:03  GET /v1/orders/10423  200  user=usr_8f21  tenant=tnt_a91  11ms

app.get('/v1/orders/:id', requireAuth, async (req, res) => {
  const order = await db.order.findUnique({ where: { id: Number(req.params.id) } })
  if (!order) return res.status(404).end()
  res.json(order)
})

// 脆弱：只凭 ID 查询
const order = await db.order.findUnique({ where: { id: orderId } })

// 安全：把归属关系放进查询条件里
const order = await db.order.findFirst({
  where: { id: orderId, tenantId: req.user.tenantId },
})
if (!order) return res.status(404).end()

// 能跑，但很容易变得脆弱的写法
const order = await db.order.findUnique({ where: { id: orderId } })
if (order.tenantId !== req.user.tenantId) return res.status(404).end()

for i in $(seq 10400 10500); do
  code=$(curl -s -o /dev/null -w '%{http_code}' \
    -H "Authorization: Bearer $TOKEN" "https://api.example.com/v1/orders/$i")
  [ "$code" = "200" ] && echo "leaked: $i"
done

from dataclasses import dataclass
from sqlalchemy.orm import Session

@dataclass(frozen=True)
class Principal:
    user_id: str
    tenant_id: str
    roles: frozenset[str]

class ScopedRepository:
    """不经过这个类就访问不到数据。"""

    def __init__(self, session: Session, principal: Principal):
        self._session = session
        self._principal = principal

    def orders(self):
        return self._session.query(Order).filter(Order.tenant_id == self._principal.tenant_id)

    def order(self, order_id: str) -> Order | None:
        return self.orders().filter(Order.id == order_id).one_or_none()

    def invoices(self):
        return self._session.query(Invoice).filter(Invoice.tenant_id == self._principal.tenant_id)

@router.get("/v1/orders/{order_id}")
def get_order(order_id: str, repo: ScopedRepository = Depends(scoped_repo)):
    order = repo.order(order_id)
    if order is None:
        raise HTTPException(status_code=404)
    return OrderOut.model_validate(order)

# .semgrep/no-unscoped-query.yaml
rules:
  - id: unscoped-orm-query
    languages: [python]
    severity: ERROR
    message: 不要直接从 session 查询，请使用 ScopedRepository。
    patterns:
      - pattern-either:
          - pattern: $SESSION.query(...)
          - pattern: $MODEL.objects.filter(...)
      - pattern-not-inside: |
          class ScopedRepository:
              ...
    paths:
      exclude:
        - tests/
        - migrations/

semgrep --config .semgrep/no-unscoped-query.yaml src/

src/api/reports.py
   57┆ rows = session.query(Order).filter(Order.status == "paid").all()
      ⚠ unscoped-orm-query
        不要直接从 session 查询，请使用 ScopedRepository。

1 finding.

package authz

import rego.v1

default allow := false

# 同一租户的资源可以读取
allow if {
    input.action == "orders:read"
    input.resource.tenant_id == input.subject.tenant_id
}

# 退款必须是同一租户，并且具备管理员角色
allow if {
    input.action == "orders:refund"
    input.resource.tenant_id == input.subject.tenant_id
    "billing_admin" in input.subject.roles
}

package authz_test

import rego.v1
import data.authz

test_cross_tenant_read_denied if {
    not authz.allow with input as {
        "action": "orders:read",
        "subject": {"tenant_id": "tnt_b04", "roles": []},
        "resource": {"tenant_id": "tnt_a91"},
    }
}

ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE orders FORCE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON orders
  USING (tenant_id = current_setting('app.tenant_id', true)::uuid)
  WITH CHECK (tenant_id = current_setting('app.tenant_id', true)::uuid);

BEGIN;
SET LOCAL app.tenant_id = '3f2b8c14-9a71-4f0e-8d33-6c2a15b7e901';
SELECT id, total_amount FROM orders WHERE id = 10422;
COMMIT;

 id | total_amount
----+--------------
(0 rows)

from sqlalchemy import event, text
from sqlalchemy.orm import Session

@event.listens_for(Session, "after_begin")
def apply_tenant_context(session, transaction, connection):
    tenant_id = current_tenant.get(None)
    if tenant_id is None:
        raise RuntimeError("没有租户上下文就不能开启事务")
    connection.execute(
        text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
        {"tenant_id": str(tenant_id)},
    )

-- 找出没有挂上策略的表
SELECT c.relname AS table_name, c.relrowsecurity, c.relforcerowsecurity
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = 'app' AND c.relkind = 'r'
ORDER BY c.relrowsecurity, c.relname;

    table_name    | relrowsecurity | relforcerowsecurity
------------------+----------------+---------------------
 audit_events     | f              | f
 webhook_deliver  | f              | f
 orders           | t              | t
 invoices         | t              | t

// 只是画面控制，并不能保证服务器会拒绝请求
{user.role === 'admin' && <DeleteButton onClick={remove} />}

curl -sS -X POST https://api.example.com/v1/users/812/role \
  -H "Authorization: Bearer $VIEWER_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"role":"admin"}' -w '\n%{http_code}\n'

{"id":812,"email":"viewer@example.com","role":"admin"}
200

// 脆弱：整个请求体都会被写入
await db.user.update({ where: { id: req.user.id }, data: req.body })

curl -sS -X PATCH https://api.example.com/v1/me \
  -H "Authorization: Bearer $VIEWER_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"displayName":"Kim","role":"admin","tenantId":"tnt_a91"}'

import { z } from 'zod'

const UpdateMe = z
  .object({
    displayName: z.string().min(1).max(80),
    locale: z.enum(['ko', 'en', 'ja']).optional(),
  })
  .strict()

app.patch('/v1/me', requireAuth, async (req, res) => {
  const parsed = UpdateMe.safeParse(req.body)
  if (!parsed.success) return res.status(400).json({ error: 'invalid payload' })
  const user = await db.user.update({ where: { id: req.user.id }, data: parsed.data })
  res.json({ id: user.id, displayName: user.displayName })
})

import pytest

# 由 fixture 填入账号 A 创建的资源标识符
CROSS_TENANT_CASES = [
    ("GET", "/v1/orders/{order_id}"),
    ("PATCH", "/v1/orders/{order_id}"),
    ("DELETE", "/v1/orders/{order_id}"),
    ("GET", "/v1/orders/{order_id}/attachments/{attachment_id}"),
    ("GET", "/v1/invoices/{invoice_id}"),
    ("POST", "/v1/invoices/{invoice_id}/refund"),
    ("GET", "/v1/webhooks/{webhook_id}"),
]

@pytest.mark.parametrize("method,template", CROSS_TENANT_CASES)
def test_tenant_b_cannot_touch_tenant_a(client, tenant_a_resources, tenant_b_auth, method, template):
    url = template.format(**tenant_a_resources)
    response = client.request(method, url, headers=tenant_b_auth, json={})
    assert response.status_code == 404, (
        f"{method} {url} 返回了 {response.status_code}。"
        f"响应体片段: {response.text[:120]}"
    )

FAILED tests/test_authz.py::test_tenant_b_cannot_touch_tenant_a[GET-/v1/orders/{order_id}/attachments/{attachment_id}]
AssertionError: GET /v1/orders/ord_5512/attachments/att_9931 返回了 200。
响应体片段: {"id":"att_9931","filename":"2026-06-invoice.pdf","url":"https://cdn...

def test_every_resource_route_has_cross_tenant_case(app):
    covered = {template for _, template in CROSS_TENANT_CASES}
    exempt = {"/health", "/v1/auth/login", "/v1/auth/refresh", "/v1/signup"}

    missing = [
        route.path
        for route in app.routes
        if "{" in route.path and route.path not in covered and route.path not in exempt
    ]
    assert not missing, f"缺少交叉租户测试的路由: {missing}"

{
  "ts": "2026-07-26T04:12:03.117Z",
  "request_id": "01J8ZQ4T7K3M9F0X2B6C5A1D8E",
  "actor_id": "usr_8f21",
  "actor_tenant": "tnt_a91",
  "action": "orders:read",
  "resource_type": "order",
  "resource_id": "ord_10423",
  "resource_tenant": "tnt_b04",
  "decision": "deny",
  "reason": "tenant_mismatch",
  "source_ip": "203.0.113.44",
  "user_agent": "python-requests/2.32.3"
}

SELECT actor_id, count(*) AS denies, count(DISTINCT resource_id) AS distinct_targets
FROM audit_events
WHERE decision = 'deny'
  AND reason = 'tenant_mismatch'
  AND ts > now() - interval '10 minutes'
GROUP BY actor_id
HAVING count(*) > 20
ORDER BY denies DESC;

 actor_id | denies | distinct_targets
----------+--------+------------------
 usr_8f21 |    317 |              317