Model Access Control

When multiple teams or customers share AI infrastructure, access control determines who can use which models, how much they can consume, and how costs are attributed. Get this wrong and one team burns through the GPU budget, or sensitive models are exposed to unauthorized users.

---

1. API Key Management

class APIKeyManager:
    async def create_key(self, tenant_id: str, scopes: list[str], 
                          expires_in_days: int = 90) -> APIKey:
        key = APIKey(
            key_id=f"sk-{uuid4().hex}",
            key_hash=bcrypt.hash(f"sk-{uuid4().hex}"),
            tenant_id=tenant_id,
            scopes=scopes,           # ["llama-70b", "llama-8b", "embedding"]
            created_at=datetime.utcnow(),
            expires_at=datetime.utcnow() + timedelta(days=expires_in_days),
            rate_limit_tokens_per_min=100_000,
            monthly_token_budget=10_000_000,
        )
        await self.db.insert(key)
        return key
    
    async def rotate_key(self, old_key_id: str) -> APIKey:
        """Create new key, mark old key for grace period expiry."""
        old_key = await self.db.get(old_key_id)
        
        # New key with same scopes
        new_key = await self.create_key(old_key.tenant_id, old_key.scopes)
        
        # Old key gets 72-hour grace period
        old_key.expires_at = datetime.utcnow() + timedelta(hours=72)
        old_key.rotated_to = new_key.key_id
        await self.db.update(old_key)
        
        return new_key

Key Scoping

Each key has granular permissions:

key_scopes = {
    "models": ["llama-70b", "llama-8b"],    # Which models can be accessed
    "operations": ["chat", "embeddings"],     # Which API endpoints
    "max_tokens_per_request": 8192,           # Per-request limit
    "rate_limit_rpm": 600,                    # Requests per minute
    "rate_limit_tpm": 100_000,                # Tokens per minute
    "ip_allowlist": ["10.0.0.0/8"],           # IP restrictions
    "valid_until": "2025-06-01T00:00:00Z",    # Expiration
}

---

2. Usage Quotas

Multi-Dimensional Quotas

class QuotaManager:
    async def check_quota(self, tenant_id: str, estimated_tokens: int) -> QuotaResult:
        usage = await self.get_usage(tenant_id)
        limits = await self.get_limits(tenant_id)
        
        checks = [
            ("tokens_per_minute", usage.tokens_this_minute, limits.tpm, estimated_tokens),
            ("requests_per_minute", usage.requests_this_minute, limits.rpm, 1),
            ("tokens_per_day", usage.tokens_today, limits.daily_tokens, estimated_tokens),
            ("tokens_per_month", usage.tokens_this_month, limits.monthly_tokens, estimated_tokens),
            ("concurrent_requests", usage.active_requests, limits.max_concurrent, 1),
        ]
        
        for name, current, limit, increment in checks:
            if current + increment > limit:
                return QuotaResult(
                    allowed=False,
                    exceeded_quota=name,
                    current=current,
                    limit=limit,
                    retry_after=self._calc_retry_after(name),
                )
        
        return QuotaResult(allowed=True)

Tier-Based Limits

Dimension	Free	Starter	Pro	Enterprise
Tokens/minute	10K	100K	1M	10M
Requests/minute	10	100	1,000	10,000
Concurrent requests	2	10	100	1,000
Monthly tokens	1M	50M	500M	Unlimited
Models available	8B only	8B, 70B	All	All + custom
Max tokens/request	2K	8K	32K	128K

---

3. Cost Attribution

Track exactly how much each tenant/team spends:

class CostTracker:
    # Cost per million tokens by model
    MODEL_COSTS = {
        "llama-8b-int4":  {"input": 0.05, "output": 0.10},
        "llama-70b-int4": {"input": 0.20, "output": 0.50},
        "llama-405b":     {"input": 1.00, "output": 3.00},
        "embedding":      {"input": 0.01, "output": 0.00},
    }
    
    async def record(self, tenant_id: str, model: str, 
                      input_tokens: int, output_tokens: int):
        costs = self.MODEL_COSTS[model]
        input_cost = input_tokens / 1_000_000 * costs["input"]
        output_cost = output_tokens / 1_000_000 * costs["output"]
        total_cost = input_cost + output_cost
        
        await self.db.insert({
            "tenant_id": tenant_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": total_cost,
            "timestamp": datetime.utcnow(),
        })

Cost Dashboards

-- Monthly cost by tenant
SELECT tenant_id, 
       sum(cost_usd) AS monthly_cost,
       sum(input_tokens + output_tokens) AS total_tokens
FROM usage_records
WHERE timestamp >= date_trunc('month', current_date)
GROUP BY tenant_id
ORDER BY monthly_cost DESC;

-- Cost by model by day (detect anomalies)
SELECT date(timestamp) AS day, model,
       sum(cost_usd) AS daily_cost
FROM usage_records
WHERE timestamp >= current_date - 30
GROUP BY day, model
ORDER BY day, model;

---

4. Model Permissions

Not all tenants should access all models:

class ModelPermissions:
    PERMISSION_MATRIX = {
        "free":       {"llama-8b"},
        "starter":    {"llama-8b", "llama-70b", "embedding"},
        "pro":        {"llama-8b", "llama-70b", "llama-405b", "embedding", "whisper"},
        "enterprise": {"*"},  # All models
    }
    
    async def check_access(self, tenant: Tenant, model: str) -> bool:
        allowed = self.PERMISSION_MATRIX.get(tenant.tier, set())
        if "*" in allowed:
            return True
        
        if model not in allowed:
            raise PermissionDenied(
                f"Model {model} not available on {tenant.tier} tier. "
                f"Available: {', '.join(sorted(allowed))}"
            )
        return True

---

5. Audit Trail

Log every API interaction for compliance and debugging:

class AuditLogger:
    async def log(self, event: AuditEvent):
        """Immutable audit log entry."""
        entry = {
            "event_id": uuid4().hex,
            "timestamp": datetime.utcnow().isoformat(),
            "tenant_id": event.tenant_id,
            "key_id": event.key_id,
            "action": event.action,       # "chat.completion", "key.created", "key.rotated"
            "model": event.model,
            "input_tokens": event.input_tokens,
            "output_tokens": event.output_tokens,
            "status": event.status,        # "success", "rate_limited", "permission_denied"
            "ip_address": event.ip_address,
            "user_agent": event.user_agent,
        }
        
        # Write to append-only storage (not deletable by tenants)
        await self.audit_store.append(entry)

Audit Queries

-- All access attempts for a specific tenant (incident investigation)
SELECT * FROM audit_log 
WHERE tenant_id = 'tenant_xyz' 
AND timestamp BETWEEN '2024-06-01' AND '2024-06-15'
ORDER BY timestamp;

-- Failed authentication attempts (security monitoring)
SELECT ip_address, count(*) AS attempts
FROM audit_log
WHERE status = 'auth_failed' AND timestamp > now() - interval '1 hour'
GROUP BY ip_address
HAVING count(*) > 10
ORDER BY attempts DESC;

---

Self-Check

• Design an API key system with scoping, rotation, and grace periods
• Implement multi-dimensional quota checking (tokens/min, requests/min, monthly budget)
• Build a cost attribution system that tracks per-tenant per-model spend
• Set up model permission tiers (free, starter, pro, enterprise)
• Design an audit trail that captures all access attempts

---

References

• OpenAI API Key Best Practices
• For rate limiting implementation, see API Gateway
• For GPU isolation between tenants, see GPU Isolation