SSO, quyền ứng dụng và quyền dữ liệu có ba hợp đồng

Câu hỏi: token hợp lệ cho biết ai đang gọi, nhưng ai quyết định người đó được đọc đơn hàng nào?

Cần biết trước: HTTP, JWT và transaction. Ví dụ là dịch vụ đơn hàng giả lập, Python 3.14.4/PyJWT 2.15.1/cryptography 50.0.2/PostgreSQL 18.6 native trên macOS arm64. Không đăng nhập IdP thật, chưa chạy discovery/JWKS rotation/revocation hoặc PKCE flow; lab chỉ kiểm phần resource server, policy và context DB.

Ba câu hỏi và các biên tin cậy

SSO dùng phiên của IdP để người dùng vào nhiều ứng dụng; OAuth/OIDC không buộc mọi API phải dùng JWT hay cấm ứng dụng giữ session. ID token dành cho client/RP kiểm kết quả đăng nhập, audience thường là client_id; access token dành cho resource server. Đừng đưa ID token vào API chỉ vì chữ ký hợp lệ. OIDC Core.

Lab chọn access token theo hợp đồng JWT cụ thể: RS256 cố định, issuer/audience tường minh, exp/iat và các claim bắt buộc, typ=at+jwt. JWT access token của hệ khác cần đối chiếu hợp đồng issuer; opaque token có đường validation khác. Không dùng URL/key/algorithm do header token tự cung cấp làm cấu hình tin cậy. RFC 9068 §4.

Backend map (iss,sub) sang user nội bộ; email/display name không là khóa identity bền. Tenant/role lấy từ membership nội bộ, không từ query parameter, header tự đặt hoặc một token role chưa thống nhất contract. UI chỉ phản ánh quyền; API phải deny.

Dựng database lab đã cô lập

Từ lab database, chép lab-common.sh, lab-local.sh, seed-pg.sql, seed-mysql.sql vào thư mục trống. Giữ nguyên guard socket/owner/cleanup; cần cả hai binary theo fixture, dù bài này chỉ query PostgreSQL. Không chạy schema dưới đây trên database của bạn. Cài dependency trong môi trường thử riêng bằng uv run --python 3.14.4 --with PyJWT==2.15.1 --with cryptography==50.0.2.

CREATE ROLE wiki_api LOGIN NOSUPERUSER NOBYPASSRLS NOINHERIT;
CREATE SCHEMA identity_lab;
CREATE TABLE identity_lab.orders (
  id integer PRIMARY KEY,
  tenant text NOT NULL CHECK (tenant IN ('alpha', 'beta')),
  status text NOT NULL
);
INSERT INTO identity_lab.orders VALUES (1,'alpha','paid'), (2,'beta','pending');
GRANT USAGE ON SCHEMA identity_lab TO wiki_api;
GRANT SELECT ON identity_lab.orders TO wiki_api;
ALTER TABLE identity_lab.orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE identity_lab.orders FORCE ROW LEVEL SECURITY;
CREATE POLICY tenant_read ON identity_lab.orders FOR SELECT TO wiki_api
USING (tenant = nullif(current_setting('app.tenant', true), ''));

Đăng nhập DB bằng wiki_api không phải owner, superuser hoặc BYPASSRLS. Thiếu tenant context thì không có hàng; chỉ grant SELECT nên DB cũng không có quyền ghi. Policy này bảo vệ query quên filter tenant. Role có thể tự đổi GUC app.tenant, nên không bảo vệ khỏi SQL injection tùy ý hoặc backend bị chiếm; tenant context phải do đường code tin cậy đặt. SET ROLE không thay permission nghiệp vụ. PostgreSQL RLS.

. ./lab-local.sh
lab_up
lab_target
lab_psql -f setup.sql
printf 'identity fixture ready\n'

API giả với mapping và quyền ở backend

from __future__ import annotations

import os
import subprocess
from dataclasses import dataclass

import jwt
from cryptography.hazmat.primitives.asymmetric import rsa

ISSUER = "https://issuer.example.org"
AUDIENCE = "orders-api"
PRIVATE = rsa.generate_private_key(public_exponent=65537, key_size=2048)
PUBLIC = PRIVATE.public_key()


@dataclass(frozen=True, slots=True)
class User:
    tenant: str
    role: str


USERS = {
    (ISSUER, "user-a"): User("alpha", "reader"),
    (ISSUER, "user-b"): User("beta", "reader"),
    (ISSUER, "user-c"): User("alpha", "none"),
}
PERMISSIONS = {"reader": frozenset({"orders:read"}), "none": frozenset()}


def pg(socket: str, script: str, tenant: str = "alpha", key: int = 1) -> list[str]:
    assert tenant in {"alpha", "beta"}
    env = {
        k: v
        for k, v in os.environ.items()
        if k not in {"PGSERVICE", "PGHOSTADDR", "PGPASSWORD"}
    }
    env.update(
        PGSERVICEFILE="/dev/null",
        PGPASSFILE=f"{socket}/no-pgpass",
        PGOPTIONS="-c client_min_messages=warning",
    )
    out = subprocess.run(
        [
            "psql",
            "-X",
            "-q",
            "-w",
            "-At",
            "-h",
            socket,
            "-p",
            "5432",
            "-U",
            "wiki_api",
            "-d",
            "postgres",
            "-v",
            "ON_ERROR_STOP=1",
            "-v",
            f"tenant={tenant}",
            "-v",
            f"key={key}",
        ],
        input=script,
        capture_output=True,
        text=True,
        env=env,
        check=True,
        timeout=15,
    )
    return out.stdout.splitlines()


def read_order(socket: str, token: str, key: object) -> tuple[int, str]:
    try:
        if jwt.get_unverified_header(token).get("typ") != "at+jwt":
            raise jwt.InvalidTokenError("wrong token kind")
        claims = jwt.decode(
            token,
            PUBLIC,
            algorithms=["RS256"],
            issuer=ISSUER,
            audience=AUDIENCE,
            options={
                "require": [
                    "iss",
                    "aud",
                    "sub",
                    "exp",
                    "iat",
                    "jti",
                    "client_id",
                    "scope",
                ]
            },
        )
        if not isinstance(claims["scope"], str):
            raise jwt.InvalidTokenError("invalid scope")
    except jwt.InvalidTokenError:
        return 401, "invalid_token"
    user = USERS.get((claims["iss"], claims["sub"]))
    if user is None or "orders:read" not in PERMISSIONS[user.role]:
        return 403, "forbidden"
    if "orders:read" not in claims["scope"].split():
        return 403, "insufficient_scope"
    if type(key) is not int:
        return 400, "invalid_id"
    rows = pg(
        socket,
        """BEGIN;
        SELECT set_config('app.tenant', :'tenant', true);
        SELECT 'ROW:'||id||':'||tenant||':'||status FROM identity_lab.orders WHERE id=:'key'::integer;
        COMMIT;""",
        tenant=user.tenant,
        key=key,
    )
    data = [line.removeprefix("ROW:") for line in rows if line.startswith("ROW:")]
    return (200, data[0]) if data else (404, "not_found")

Hai lớp kiểm quyền đều ở backend: role cho phép đọc và token có scope đọc. Query chỉ filter ID để RLS chịu trách nhiệm tenant. ID không phải integer bị trả 400 ngay ở handler; bool cũng không được xem là ID. :'key'::integer quote giá trị theo cú pháp psql rồi ép kiểu SQL, thay vì phép thay thế thô :key. Tenant được quote tương tự. Response404 che hàng không thuộc tenant; lab không có HTTP transport, status/body là kết quả handler giả.

Token sai, quyền thiếu và connection được dùng lại

from __future__ import annotations

import sys
import time
from importlib.metadata import version

import jwt
from auth import AUDIENCE, ISSUER, PRIVATE, pg, read_order
from cryptography.hazmat.primitives.asymmetric import rsa

assert sys.version_info[:3] == (3, 14, 4)
assert version("PyJWT") == "2.15.1" and version("cryptography") == "50.0.2"
socket = sys.argv[1]
now = int(time.time())
payload: dict[str, object] = {
    "iss": ISSUER,
    "aud": AUDIENCE,
    "sub": "user-a",
    "iat": now,
    "exp": now + 300,
    "jti": "fake-request",
    "client_id": "orders-web",
    "scope": "orders:read",
}


def issue(changes: dict[str, object] | None = None, typ: str = "at+jwt") -> str:
    return jwt.encode(
        payload | (changes or {}), PRIVATE, algorithm="RS256", headers={"typ": typ}
    )


for label, token in [
    ("expired", issue({"iat": now - 100, "exp": now - 10})),
    ("audience", issue({"aud": "other-api"})),
    ("issuer", issue({"iss": "untrusted-issuer"})),
    ("id_token", issue({"aud": "orders-web"}, typ="JWT")),
    ("wrong_typ_same_aud", issue(typ="JWT")),
    ("future", issue({"nbf": now + 300})),
    (
        "wrong_key",
        jwt.encode(
            payload,
            rsa.generate_private_key(public_exponent=65537, key_size=2048),
            algorithm="RS256",
            headers={"typ": "at+jwt"},
        ),
    ),
    (
        "wrong_alg",
        jwt.encode(
            payload,
            b"fake-test-key-never-use-in-production",
            algorithm="HS256",
            headers={"typ": "at+jwt"},
        ),
    ),
    ("malformed", "invalid"),
]:
    assert read_order(socket, token, 1) == (401, "invalid_token"), label
    print(label, "401")
missing = payload.copy()
del missing["exp"]
assert (
    read_order(
        socket,
        jwt.encode(missing, PRIVATE, algorithm="RS256", headers={"typ": "at+jwt"}),
        1,
    )[0]
    == 401
)
assert read_order(socket, issue({"sub": "user-c"}), 1) == (403, "forbidden")
assert read_order(socket, issue({"sub": "unknown"}), 1) == (403, "forbidden")
assert read_order(socket, issue({"scope": "orders:write"}), 1) == (
    403,
    "insufficient_scope",
)
assert read_order(socket, issue(), 1) == (200, "1:alpha:paid")
assert read_order(socket, issue(), 2) == (404, "not_found")
assert read_order(socket, issue({"sub": "user-b"}), 2) == (200, "2:beta:pending")
assert read_order(socket, issue({"sub": "user-b"}), 1) == (404, "not_found")
print("missing claim401; no permission403; scope403; tenants200/404")

for invalid_key in [
    (
        "0; SELECT set_config('app.tenant','beta',true); "
        "SELECT 'ROW:'||id||':'||tenant||':'||status FROM identity_lab.orders WHERE id=2"
    ),
    True,
    1.5,
]:
    result = read_order(socket, issue(), invalid_key)
    assert result == (400, "invalid_id"), result
print("untrusted id400; tenant-switch injection rejected")

lines = pg(
    socket,
    """
SELECT 'VERSION:'||current_setting('server_version_num');
SELECT 'ROLE:'||current_user||':'||rolsuper||':'||rolbypassrls FROM pg_roles WHERE rolname=current_user;
SELECT 'PID:'||pg_backend_pid();
SELECT 'EMPTY:'||count(*) FROM identity_lab.orders;
BEGIN;
SELECT set_config('app.tenant','alpha',true);
SELECT 'A:'||count(*) FROM identity_lab.orders;
SELECT 'A_FOREIGN:'||count(*) FROM identity_lab.orders WHERE id=2;
COMMIT;
SELECT 'PID:'||pg_backend_pid();
SELECT 'CLEAR_COMMIT:'||count(*) FROM identity_lab.orders;
BEGIN;
SELECT set_config('app.tenant','beta',true);
SELECT 'B:'||count(*) FROM identity_lab.orders;
SELECT 'B_FOREIGN:'||count(*) FROM identity_lab.orders WHERE id=1;
ROLLBACK;
SELECT 'PID:'||pg_backend_pid();
SELECT 'CLEAR_ROLLBACK:'||count(*) FROM identity_lab.orders;
""",
)
pids = [line for line in lines if line.startswith("PID:")]
assert len(pids) == 3 and len(set(pids)) == 1
assert {
    "VERSION:180006",
    "ROLE:wiki_api:false:false",
    "EMPTY:0",
    "A:1",
    "A_FOREIGN:0",
    "CLEAR_COMMIT:0",
    "B:1",
    "B_FOREIGN:0",
    "CLEAR_ROLLBACK:0",
} <= set(lines), lines
print("RLS same connection A1/B1 foreign0; commit/rollback context0")

. ./lab.env
uv run --python 3.14.4 --with PyJWT==2.15.1 --with cryptography==50.0.2 python check.py "$LAB_DIR"

Expected: token sai401, quyền thiếu403, own tenant200/cross-tenant404; query trực tiếp bỏ WHEREtenant vẫn chỉ thấy một hàng. Ba PID phải cùng một connection, context rỗng sau commit và rollback. Đây là mô phỏng tái dùng connection, chưa kiểm PgBouncer hoặc driver pool. SET LOCAL hết ở cuối transaction; session SET có thể rò context sang request sau. Pooling contracts.

. ./lab-local.sh
lab_clean
lab_clean
test ! -f lab.env

Điều lab chưa bảo đảm

RSA key sinh riêng mỗi process, không lưu/in token/key. Production cần key lấy từ issuer đã tin, TLS/discovery, rotation/cache failure và giới hạn lifetime/revocation đã kiểm. Token đúng signature/expiry không chứng minh user vẫn còn membership; lab dùng mapping backend hiện tại, không đo propagation/identity lifecycle.

Policy DB không hiểu scope hoặc permission endpoint; role đọc trong tenant không được tự suy thành quyền hoàn tiền. API vẫn phải kiểm nghiệp vụ. Superuser/BYPASSRLS và một số table-owner path có thể vượt RLS; FORCE không thắng superuser. Custom GUC có thể đổi bởi SQL, nên dùng principal tin cậy ở biên context và kiểm injection riêng. Lab chỉ SELECT hai hàng, không có write policy, audit durable hoặc chuẩn bảo mật đã chứng nhận. Học tiếp: review AI code, ADR, MVCC.