| name | security-compliance |
| description | Skill for ensuring security best practices and regulatory compliance. Use when implementing authentication, handling sensitive data, securing APIs, or meeting compliance requirements (PCI-DSS, GDPR). Provides OWASP Top 10 prevention patterns, security headers, encryption strategies, and data protection guidelines. |
Security Compliance
Skill for implementing security best practices and meeting compliance requirements.
Overview
This skill provides guidance for:
- OWASP Top 10 - Prevention of common vulnerabilities
- Authentication - Secure auth implementation patterns
- Data Protection - Encryption and handling of sensitive data
- Compliance - PCI-DSS, GDPR requirements
- Security Headers - CSP, HSTS, and other protective headers
OWASP Top 10 Prevention
A01: Broken Access Control
Vulnerability: Users can act outside intended permissions.
Prevention:
from fastapi import Depends, HTTPException, status
async def require_permission(permission: str):
async def checker(user: User = Depends(get_current_user)):
if not user.has_permission(permission):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Insufficient permissions"
)
return user
return checker
@router.delete("/users/{user_id}")
async def delete_user(
user_id: int,
current_user: User = Depends(require_permission("admin:delete_users"))
):
await user_service.delete(user_id)
import { NextResponse } from 'next/server';
import type { NextRequest } from 'next/server';
export function middleware(request: NextRequest) {
const token = request.cookies.get('session')?.value;
if (request.nextUrl.pathname.startsWith('/admin')) {
if (!token) {
return NextResponse.redirect(new URL('/login', request.url));
}
const payload = verifyToken(token);
if (payload.role !== 'admin') {
return NextResponse.redirect(new URL('/unauthorized', request.url));
}
}
return NextResponse.next();
}
A02: Cryptographic Failures
Vulnerability: Sensitive data exposed due to weak/missing encryption.
Prevention:
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class AuthService:
def hash_password(self, password: str) -> str:
return pwd_context.hash(password)
def verify_password(self, plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
from cryptography.fernet import Fernet
import base64
import os
class FieldEncryption:
def __init__(self):
key = os.environ.get("ENCRYPTION_KEY")
self.cipher = Fernet(key.encode())
def encrypt(self, value: str) -> str:
return self.cipher.encrypt(value.encode()).decode()
def decrypt(self, encrypted: str) -> str:
return self.cipher.decrypt(encrypted.encode()).decode()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String, nullable=False)
_ssn_encrypted = Column("ssn", String)
@property
def ssn(self):
if self._ssn_encrypted:
return encryption.decrypt(self._ssn_encrypted)
return None
@ssn.setter
():
._ssn_encrypted = encryption.encrypt(value)
A03: Injection
Vulnerability: SQL, NoSQL, OS command injection.
Prevention:
from sqlalchemy import text
query = f"SELECT * FROM users WHERE email = '{email}'"
result = session.execute(
text("SELECT * FROM users WHERE email = :email"),
{"email": email}
)
user = session.query(User).filter(User.email == email).first()
from pydantic import BaseModel, EmailStr, constr, validator
import re
class UserCreate(BaseModel):
email: EmailStr
username: constr(min_length=3, max_length=50, pattern=r'^[a-zA-Z0-9_]+$')
bio: str
@validator('bio')
def sanitize_bio(cls, v):
return re.sub(r'<script[^>]*>.*?</script>', '', v, flags=re.IGNORECASE)
A04: Insecure Design
Vulnerability: Missing security controls in design phase.
Prevention Checklist:
A05: Security Misconfiguration
Vulnerability: Default configs, unnecessary features, missing patches.
Prevention:
DEBUG = False
CORS_ORIGINS = ["https://yourdomain.com"]
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = "Strict"
DOCS_URL = None
REDOC_URL = None
module.exports = {
poweredBy: false,
reactStrictMode: true,
async headers() {
return [
{
source: '/:path*',
headers: securityHeaders,
},
];
},
};
A06: Vulnerable and Outdated Components
Prevention:
npm audit
pip-audit
npm outdated
pip list --outdated
npm ci
pip install -r requirements.txt --require-hashes
A07: Identification and Authentication Failures
Prevention:
from fastapi import FastAPI, Response
from datetime import timedelta
SESSION_SETTINGS = {
"secret_key": os.environ["SESSION_SECRET"],
"expire_after": timedelta(hours=8),
"secure": True,
"httponly": True,
"samesite": "strict",
}
class LoginService:
MAX_ATTEMPTS = 5
LOCKOUT_DURATION = timedelta(minutes=15)
async def login(self, email: str, password: str) -> User:
attempts = await self.get_failed_attempts(email)
if attempts >= self.MAX_ATTEMPTS:
lockout_expires = await self.get_lockout_expiry(email)
if datetime.utcnow() < lockout_expires:
raise HTTPException(
status_code=429,
detail=f"Account locked. Try again in {lockout_expires - datetime.utcnow()}"
)
user = await self.verify_credentials(email, password)
if not user:
.record_failed_attempt(email)
HTTPException(status_code=, detail=)
.clear_failed_attempts(email)
user
A08: Software and Data Integrity Failures
Prevention:
import hmac
import hashlib
def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
expected = hmac.new(
secret.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
@router.post("/webhooks/payment")
async def handle_payment_webhook(request: Request):
payload = await request.body()
signature = request.headers.get("X-Signature")
if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
raise HTTPException(status_code=401, detail="Invalid signature")
A09: Security Logging and Monitoring Failures
Prevention:
import structlog
from datetime import datetime
logger = structlog.get_logger()
class SecurityAuditLogger:
async def log_auth_event(
self,
event_type: str,
user_id: str | None,
ip_address: str,
success: bool,
details: dict = None
):
await logger.ainfo(
"security_event",
event_type=event_type,
user_id=user_id,
ip_address=ip_address,
success=success,
timestamp=datetime.utcnow().isoformat(),
details=details or {}
)
audit = SecurityAuditLogger()
@router.post("/login")
async def login(request: Request, credentials: LoginRequest):
ip = request.client.host
try:
user = await auth_service.login(credentials)
await audit.log_auth_event("login", user.id, ip, success=True)
return {"token": create_token(user)}
except AuthenticationError:
await audit.log_auth_event(
"login",
None,
ip,
success=False,
details={"email": credentials.email}
)
A10: Server-Side Request Forgery (SSRF)
Prevention:
from urllib.parse import urlparse
import ipaddress
ALLOWED_HOSTS = ["api.example.com", "cdn.example.com"]
def validate_url(url: str) -> bool:
parsed = urlparse(url)
if parsed.scheme != "https":
return False
if parsed.hostname not in ALLOWED_HOSTS:
return False
try:
ip = ipaddress.ip_address(parsed.hostname)
if ip.is_private or ip.is_loopback:
return False
except ValueError:
pass
return True
@router.post("/fetch-preview")
async def fetch_url_preview(url: str):
if not validate_url(url):
raise HTTPException(status_code=400, detail="Invalid URL")
response = await httpx.get(url, timeout=)
parse_preview(response)
Security Headers
Implementation
const securityHeaders = [
{
key: 'X-DNS-Prefetch-Control',
value: 'on'
},
{
key: 'Strict-Transport-Security',
value: 'max-age=63072000; includeSubDomains; preload'
},
{
key: 'X-Frame-Options',
value: 'SAMEORIGIN'
},
{
key: 'X-Content-Type-Options',
value: 'nosniff'
},
{
key: 'X-XSS-Protection',
value: '1; mode=block'
},
{
key: 'Referrer-Policy',
value: 'strict-origin-when-cross-origin'
},
{
key: 'Permissions-Policy',
value: 'camera=(), microphone=(), geolocation=()'
},
{
key: 'Content-Security-Policy',
value: ContentSecurityPolicy.replace(/\s{2,}/g, ' ').trim()
}
];
const ContentSecurityPolicy = `
default-src 'self';
script-src 'self' 'unsafe-eval' 'unsafe-inline';
style-src 'self' 'unsafe-inline';
img-src 'self' blob: data: https:;
font-src 'self';
connect-src 'self' https://api.yourdomain.com;
frame-ancestors 'none';
base-uri 'self';
form-action 'self';
`;
Header Reference
| Header | Purpose | Value |
|---|
| Strict-Transport-Security | Force HTTPS | max-age=31536000; includeSubDomains |
| X-Frame-Options | Prevent clickjacking | DENY or SAMEORIGIN |
| X-Content-Type-Options | Prevent MIME sniffing | nosniff |
| Content-Security-Policy | Control resource loading | See CSP guide |
| X-XSS-Protection | Legacy XSS filter | 1; mode=block |
| Referrer-Policy | Control referrer info | strict-origin-when-cross-origin |
| Permissions-Policy | Disable browser features | camera=(), microphone=() |
PCI-DSS Compliance
Requirements Overview
| Requirement | Implementation |
|---|
| Protect cardholder data | Encryption at rest and in transit |
| Maintain vulnerability management | Regular patching, security scanning |
| Implement strong access control | RBAC, MFA, audit logging |
| Monitor and test networks | IDS/IPS, penetration testing |
| Maintain security policy | Documented procedures |
Payment Data Handling
class PaymentService:
def __init__(self, payment_provider):
self.provider = payment_provider
async def process_payment(
self,
amount: int,
card_token: str,
user_id: str
) -> PaymentResult:
result = await self.provider.charge(
amount=amount,
source=card_token,
metadata={"user_id": user_id}
)
await self.store_transaction(
user_id=user_id,
provider_id=result.id,
last_four=result.card.last4,
amount=amount,
status=result.status
)
return result
import { loadStripe } from '@stripe/stripe-js';
import { Elements, CardElement, useStripe } from '@stripe/react-stripe-js';
function PaymentForm() {
const stripe = useStripe();
const handleSubmit = async (e: FormEvent) => {
e.preventDefault();
const { token, error } = await stripe.createToken(cardElement);
if (token) {
await api.processPayment({ token: token.id, amount });
}
};
return (
<form onSubmit={handleSubmit}>
<CardElement />
<button type="submit">Pay</button>
</form>
);
}
GDPR Compliance
Data Subject Rights
| Right | Implementation |
|---|
| Access | Export user data endpoint |
| Rectification | Update profile endpoint |
| Erasure | Delete account endpoint |
| Portability | JSON/CSV export |
| Object | Opt-out mechanisms |
Implementation
class GDPRService:
async def export_user_data(self, user_id: str) -> dict:
"""Export all user data for GDPR access request."""
user = await user_repo.get(user_id)
return {
"profile": {
"email": user.email,
"name": user.name,
"created_at": user.created_at.isoformat(),
},
"orders": await order_repo.get_by_user(user_id),
"activity_log": await activity_repo.get_by_user(user_id),
"preferences": await preference_repo.get_by_user(user_id),
}
async def delete_user_data(self, user_id: str) -> None:
"""Delete all user data for GDPR erasure request."""
await user_repo.anonymize(user_id)
await order_repo.anonymize_user(user_id)
await activity_repo.delete_by_user(user_id)
await audit_log.record(
event="gdpr_erasure",
user_id=user_id,
timestamp=datetime.utcnow()
)
Consent Management
interface ConsentState {
necessary: true;
analytics: boolean;
marketing: boolean;
}
function CookieConsent() {
const [consent, setConsent] = useState<ConsentState | null>(null);
const handleAccept = (options: Partial<ConsentState>) => {
const newConsent = { necessary: true, ...options };
setConsent(newConsent);
setCookie('consent', JSON.stringify(newConsent), { maxAge: 365 * 24 * 60 * 60 });
if (newConsent.analytics) {
initAnalytics();
}
};
return (
<div className="cookie-banner">
<p>We use cookies to improve your experience.</p>
<button = => handleAccept({ analytics: true, marketing: true })}>
Accept All
handleAccept({ analytics: false, marketing: false })}>
Essential Only
);
}
Encryption Best Practices
At Rest
from sqlalchemy_utils import EncryptedType
from sqlalchemy_utils.types.encrypted.encrypted_type import AesEngine
class SensitiveData(Base):
__tablename__ = "sensitive_data"
id = Column(Integer, primary_key=True)
ssn = Column(EncryptedType(
String,
os.environ["DB_ENCRYPTION_KEY"],
AesEngine,
"pkcs5"
))
In Transit
from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)
DATABASE_URL = (
"postgresql://user:pass@host/db"
"?sslmode=require"
"&sslrootcert=/path/to/ca.pem"
)
Security Checklist
Authentication
Data Protection
API Security
Infrastructure
Compliance
References
For detailed guidance, see:
references/owasp-prevention.md - Detailed OWASP prevention strategies
references/compliance-checklist.md - Comprehensive compliance requirements