Implementing authorization logic in APIs
Scope: Authorization models, permission systems, role-based access control Lines: ~240 Last Updated: 2025-10-18
Activate this skill when:
Authentication: "Who are you?" - Verifying identity (login, JWT, OAuth) Authorization: "What can you do?" - Verifying permissions
This skill focuses on authorization only. For authentication patterns, see api-authentication.md.
What: Direct mapping of users/groups to resources and permissions.
Structure:
Resource → List of (User/Group, Permissions)
Example:
{
"document_123": {
"alice": ["read", "write"],
"bob": ["read"],
"editors_group": ["read", "write", "delete"]
}
}
Strengths:
Weaknesses:
Best for:
What: Permissions assigned to roles, users assigned to roles.
Structure:
User → Role → Permissions
Example:
{
"roles": {
"admin": ["users:read", "users:write", "users:delete", "posts:*"],
"editor": ["posts:read", "posts:write", "posts:delete"],
"viewer": ["posts:read"]
},
"users": {
"alice": ["admin"],
"bob": ["editor"],
"charlie": ["viewer"]
}
}
Strengths:
Weaknesses:
Best for:
What: Permissions based on attributes (user, resource, environment).
Structure:
Policy: IF (user attributes + resource attributes + environment) THEN allow/deny
Example:
# Policy: Users can edit their own posts during business hours
if (
user.id == post.author_id
and current_time >= "09:00"
and current_time <= "17:00"
and user.department == post.department
):
allow("edit")
Attributes:
Strengths:
Weaknesses:
Best for:
Start: What's your primary requirement?
│
├─ Simple resource sharing?
│ └─ ACL (Google Docs-style)
│
├─ Predictable organizational roles?
│ ├─ Few roles (<10) → RBAC
│ ├─ Many roles (>20) → Continue
│ └─ Complex hierarchy → Hierarchical RBAC
│
├─ Context-dependent access? (time, location, attributes)
│ └─ ABAC (with policy engine)
│
├─ Resource-level permissions needed?
│ ├─ Per-user → ACL or ABAC
│ ├─ Per-role → RBAC with resource scopes
│ └─ Dynamic → ABAC
│
├─ Audit requirements?
│ ├─ "Who has access?" → ACL or RBAC
│ └─ "Who accessed X?" → Any (with logging)
│
└─ Default → RBAC (most common)
| Criteria | RBAC | ABAC | |----------|------|------| | Complexity | Low (roles + permissions) | High (policies + attributes) | | Granularity | Coarse (role-level) | Fine (attribute-level) | | Scalability | Good (role explosion risk) | Excellent (policy-based) | | Performance | Fast (lookup) | Slower (policy evaluation) | | Auditability | Easy ("list role permissions") | Hard ("evaluate all policies") | | Context-awareness | None | Full (time, location, etc.) | | Implementation time | Days | Weeks/months | | Best for | 80% of applications | Complex compliance scenarios |
-- Users table
CREATE TABLE users (
id SERIAL PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL
);
-- Roles table
CREATE TABLE roles (
id SERIAL PRIMARY KEY,
name VARCHAR(50) UNIQUE NOT NULL, -- e.g., "admin", "editor"
description TEXT
);
-- Permissions table
CREATE TABLE permissions (
id SERIAL PRIMARY KEY,
resource VARCHAR(100) NOT NULL, -- e.g., "posts", "users"
action VARCHAR(50) NOT NULL, -- e.g., "read", "write", "delete"
UNIQUE(resource, action)
);
-- Role-Permission mapping (many-to-many)
CREATE TABLE role_permissions (
role_id INT REFERENCES roles(id),
permission_id INT REFERENCES permissions(id),
PRIMARY KEY (role_id, permission_id)
);
-- User-Role mapping (many-to-many)
CREATE TABLE user_roles (
user_id INT REFERENCES users(id),
role_id INT REFERENCES roles(id),
PRIMARY KEY (user_id, role_id)
);
-- Check if user has permission
SELECT EXISTS (
SELECT 1
FROM user_roles ur
JOIN role_permissions rp ON ur.role_id = rp.role_id
JOIN permissions p ON rp.permission_id = p.id
WHERE ur.user_id = $1 -- User ID
AND p.resource = $2 -- e.g., "posts"
AND p.action = $3 -- e.g., "write"
) AS has_permission;
from functools import wraps
from fastapi import HTTPException, Depends
def require_permission(resource: str, action: str):
"""Decorator to check permissions"""
def decorator(func):
@wraps(func)
async def wrapper(*args, current_user=Depends(get_current_user), **kwargs):
# Query database for permission
has_perm = await db.fetch_one(
"""
SELECT EXISTS (
SELECT 1
FROM user_roles ur
JOIN role_permissions rp ON ur.role_id = rp.role_id
JOIN permissions p ON rp.permission_id = p.id
WHERE ur.user_id = $1 AND p.resource = $2 AND p.action = $3
) AS has_permission
""",
[current_user.id, resource, action]
)
if not has_perm["has_permission"]:
raise HTTPException(status_code=403, detail="Permission denied")
return await func(*args, current_user=current_user, **kwargs)
return wrapper
return decorator
# Usage
@app.delete("/posts/{post_id}")
@require_permission("posts", "delete")
async def delete_post(post_id: int, current_user: User):
# Delete post logic
pass
Scenario: Editors can edit posts, but only their own posts.
Solution 1: Hybrid RBAC + Resource Check
@app.put("/posts/{post_id}")
@require_permission("posts", "write")
async def update_post(post_id: int, current_user: User):
post = await db.fetch_one("SELECT * FROM posts WHERE id = $1", [post_id])
if not post:
raise HTTPException(status_code=404)
# Resource-level check
if post["author_id"] != current_user.id and "admin" not in current_user.roles:
raise HTTPException(status_code=403, detail="Not your post")
# Update post
pass
Solution 2: Scope-Based Permissions
-- Add scope to permissions
ALTER TABLE role_permissions ADD COLUMN scope VARCHAR(50);
-- scope values: "all", "own", "department"
-- Check with scope
SELECT
p.action,
rp.scope
FROM user_roles ur
JOIN role_permissions rp ON ur.role_id = rp.role_id
JOIN permissions p ON rp.permission_id = p.id
WHERE ur.user_id = $1
AND p.resource = $2;
def check_resource_permission(user, resource, action, resource_obj):
perms = get_user_permissions(user, resource, action)
for perm in perms:
if perm.scope == "all":
return True
elif perm.scope == "own" and resource_obj.owner_id == user.id:
return True
elif perm.scope == "department" and resource_obj.department == user.department:
return True
return False
Example: Manager inherits all Employee permissions + additional permissions.
-- Add parent_role_id for hierarchy
ALTER TABLE roles ADD COLUMN parent_role_id INT REFERENCES roles(id);
-- Example hierarchy
INSERT INTO roles (name, parent_role_id) VALUES
('employee', NULL),
('manager', (SELECT id FROM roles WHERE name = 'employee')),
('director', (SELECT id FROM roles WHERE name = 'manager'));
Permission check with inheritance:
-- Recursive CTE to get all inherited permissions
WITH RECURSIVE role_hierarchy AS (
-- Base: User's direct roles
SELECT r.id, r.name, r.parent_role_id
FROM user_roles ur
JOIN roles r ON ur.role_id = r.id
WHERE ur.user_id = $1
UNION
-- Recursive: Parent roles
SELECT r.id, r.name, r.parent_role_id
FROM roles r
JOIN role_hierarchy rh ON r.id = rh.parent_role_id
)
SELECT EXISTS (
SELECT 1
FROM role_hierarchy rh
JOIN role_permissions rp ON rh.id = rp.role_id
JOIN permissions p ON rp.permission_id = p.id
WHERE p.resource = $2 AND p.action = $3
) AS has_permission;
What: Declarative policy engine using Rego language.
Use case: Complex ABAC policies, Kubernetes admission control.
Example policy (policy.rego):
package app.authz
# Allow if user is admin
allow {
input.user.role == "admin"
}
# Allow if user owns the resource
allow {
input.user.id == input.resource.owner_id
}
# Allow if user is in same department and resource is not confidential
allow {
input.user.department == input.resource.department
input.resource.confidential == false
}
Usage (Python):
import requests
def check_permission(user, resource, action):
policy_input = {
"user": {
"id": user.id,
"role": user.role,
"department": user.department
},
"resource": {
"id": resource.id,
"owner_id": resource.owner_id,
"department": resource.department,
"confidential": resource.confidential
},
"action": action
}
response = requests.post(
"http://opa:8181/v1/data/app/authz/allow",
json={"input": policy_input}
)
return response.json().get("result", False)
Pros:
Cons:
What: Authorization library with multiple models (RBAC, ABAC, ACL).
Use case: Embed authorization in application code.
Example (Python):
import casbin
# Load model and policy
enforcer = casbin.Enforcer("model.conf", "policy.csv")
# Check permission
if enforcer.enforce("alice", "posts", "write"):
print("Allowed")
else:
print("Denied")
Model file (model.conf):
[request_definition]
r = sub, obj, act
[policy_definition]
p = sub, obj, act
[role_definition]
g = _, _
[policy_effect]
e = some(where (p.eft == allow))
[matchers]
m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act
Policy file (policy.csv):
p, admin, posts, *
p, editor, posts, write
p, editor, posts, read
g, alice, admin
g, bob, editor
Pros:
Cons:
What: Embed permissions in JWT token.
Example:
{
"sub": "alice",
"role": "editor",
"scopes": ["posts:read", "posts:write", "users:read"]
}
Authorization middleware:
def require_scope(required_scope: str):
def decorator(func):
@wraps(func)
async def wrapper(*args, token=Depends(get_token), **kwargs):
scopes = token.get("scopes", [])
if required_scope not in scopes:
raise HTTPException(status_code=403)
return await func(*args, **kwargs)
return wrapper
return decorator
@app.delete("/posts/{post_id}")
@require_scope("posts:delete")
async def delete_post(post_id: int):
pass
Pros:
Cons:
Problem: Implementing authentication but forgetting authorization.
# ❌ BAD: Only checks authentication
@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, current_user=Depends(get_current_user)):
await db.execute("DELETE FROM posts WHERE id = $1", [post_id])
Solution: Always check ownership or permissions.
# ✅ GOOD: Checks both authentication and authorization
@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, current_user=Depends(get_current_user)):
post = await db.fetch_one("SELECT * FROM posts WHERE id = $1", [post_id])
if not post:
raise HTTPException(status_code=404)
if post["author_id"] != current_user.id and "admin" not in current_user.roles:
raise HTTPException(status_code=403)
await db.execute("DELETE FROM posts WHERE id = $1", [post_id])
Problem: Using predictable IDs without authorization.
# ❌ BAD: Anyone can access any invoice
@app.get("/invoices/{invoice_id}")
async def get_invoice(invoice_id: int, current_user=Depends(get_current_user)):
return await db.fetch_one("SELECT * FROM invoices WHERE id = $1", [invoice_id])
Solution: Check ownership.
# ✅ GOOD: Only owner can access invoice
@app.get("/invoices/{invoice_id}")
async def get_invoice(invoice_id: int, current_user=Depends(get_current_user)):
invoice = await db.fetch_one(
"SELECT * FROM invoices WHERE id = $1 AND user_id = $2",
[invoice_id, current_user.id]
)
if not invoice:
raise HTTPException(status_code=404) # Don't leak existence
return invoice
Problem: Users can assign themselves higher roles.
# ❌ BAD: Users can make themselves admin
@app.put("/users/{user_id}/role")
async def update_role(user_id: int, role: str, current_user=Depends(get_current_user)):
await db.execute("UPDATE users SET role = $1 WHERE id = $2", [role, user_id])
Solution: Only admins can assign roles.
# ✅ GOOD: Only admins can assign roles
@app.put("/users/{user_id}/role")
@require_permission("users", "manage_roles")
async def update_role(user_id: int, role: str, current_user=Depends(get_current_user)):
await db.execute("UPDATE users SET role = $1 WHERE id = $2", [role, user_id])
Problem: Permissions cached in JWT or session, changes not reflected.
Solution:
# Hybrid approach: Cache non-critical, query critical
if action in ["delete", "admin_access"]:
# Always query database for critical actions
has_perm = await db.fetch_one(query)
else:
# Use cached permissions from token
has_perm = action in token["scopes"]
✅ Always check authorization (not just authentication) ✅ Fail closed (deny by default, explicit allow) ✅ Check ownership for resource-level permissions ✅ Log authorization failures (audit trail) ✅ Use principle of least privilege (minimal permissions) ✅ Test with different roles (ensure isolation)
❌ Don't trust client-side authorization (always check server-side) ❌ Don't leak information (404 instead of 403 for private resources) ❌ Don't use GET for permission changes (CSRF risk) ❌ Don't cache permissions indefinitely (use short TTL) ❌ Don't expose internal role names (use display names)
[ ] Authentication verified (who is the user?)
[ ] Authorization verified (what can they do?)
[ ] Ownership checked (if resource-level)
[ ] Role/permissions cached (with TTL)
[ ] Authorization failures logged
[ ] Edge cases tested (no role, multiple roles, escalation)
[ ] IDOR vulnerabilities checked
[ ] Privilege escalation prevented
[ ] Database schema (users, roles, permissions, mappings)
[ ] Permission check function/query
[ ] Authorization middleware/decorator
[ ] Resource-level checks (ownership)
[ ] Role hierarchy (if needed)
[ ] Audit logging
[ ] Unit tests (permission matrix)
api-authentication.md - JWT, OAuth, session managementdatabase-security.md - Row-level security, SQL injection preventionapi-rate-limiting.md - Rate limiting per role/userapi-multi-tenancy.md - Tenant isolation in authorizationLast Updated: 2025-10-18 Format Version: 1.0 (Atomic)