Add authentication schemas for user registration and login: - UserRegister: email, password (with strength validation), password_confirm - UserLogin: email, password - UserResponse: id, email, created_at, is_active (orm_mode=True) - TokenResponse: access_token, token_type, expires_in - TokenData: user_id, exp Includes field validators for password strength and password confirmation matching. Test coverage: 19 tests for all schemas
17 KiB
17 KiB
Prompt: User Authentication Implementation (T17-T22)
🎯 OBIETTIVO
Implementare la fase User Authentication del progetto OpenRouter API Key Monitor seguendo rigorosamente TDD (Test-Driven Development).
Task da completare: T17, T18, T19, T20, T21, T22
📋 CONTESTO
- Repository:
/home/google/Sources/LucaSacchiNet/openrouter-watcher - Specifiche:
/home/google/Sources/LucaSacchiNet/openrouter-watcher/export/architecture.md(sezioni 4, 5) - Kanban:
/home/google/Sources/LucaSacchiNet/openrouter-watcher/export/kanban.md - Stato Attuale: Security Services completati (T01-T16), 202 test passanti
- Progresso: 22% (16/74 task)
- Servizi Pronti: Encryption, Password hashing, JWT, API Token
🔧 TASK DETTAGLIATI
T17: Creare Pydantic Schemas per Autenticazione
Requisiti:
- Creare
src/openrouter_monitor/schemas/auth.py - Schemas per request/response:
UserRegister: email, password, password_confirmUserLogin: email, passwordUserResponse: id, email, created_at, is_activeTokenResponse: access_token, token_type, expires_inTokenData: user_id (sub), exp
- Validazione password strength (richiama
validate_password_strength) - Validazione email formato valido
- Validazione password e password_confirm coincidono
Implementazione:
from pydantic import BaseModel, EmailStr, Field, validator, root_validator
class UserRegister(BaseModel):
email: EmailStr
password: str = Field(..., min_length=12, max_length=128)
password_confirm: str
@validator('password')
def password_strength(cls, v):
from openrouter_monitor.services.password import validate_password_strength
if not validate_password_strength(v):
raise ValueError('Password does not meet strength requirements')
return v
@root_validator
def passwords_match(cls, values):
if values.get('password') != values.get('password_confirm'):
raise ValueError('Passwords do not match')
return values
class UserLogin(BaseModel):
email: EmailStr
password: str
class UserResponse(BaseModel):
id: int
email: str
created_at: datetime
is_active: bool
class Config:
orm_mode = True
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
expires_in: int
class TokenData(BaseModel):
user_id: int | None = None
Test richiesti:
- Test UserRegister valido
- Test UserRegister password troppo corta
- Test UserRegister password e confirm non coincidono
- Test UserRegister email invalida
- Test UserLogin valido
- Test UserResponse orm_mode
T18: Implementare Endpoint POST /api/auth/register
Requisiti:
- Creare
src/openrouter_monitor/routers/auth.py - Endpoint:
POST /api/auth/register - Riceve
UserRegisterschema - Verifica email non esista già nel DB
- Hash password con
hash_password() - Crea utente nel database
- Ritorna
UserResponsecon status 201 - Gestire errori: email esistente (400), validazione fallita (422)
Implementazione:
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
router = APIRouter(prefix="/api/auth", tags=["auth"])
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserRegister, db: Session = Depends(get_db)):
# Verifica email esistente
existing = db.query(User).filter(User.email == user_data.email).first()
if existing:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# Crea utente
hashed_password = hash_password(user_data.password)
user = User(email=user_data.email, password_hash=hashed_password)
db.add(user)
db.commit()
db.refresh(user)
return user
Test richiesti:
- Test registrazione nuovo utente successo
- Test registrazione email esistente fallisce
- Test registrazione password debole fallisce
- Test registrazione email invalida fallisce
T19: Implementare Endpoint POST /api/auth/login
Requisiti:
- Endpoint:
POST /api/auth/login - Riceve
UserLoginschema - Verifica esistenza utente per email
- Verifica password con
verify_password() - Genera JWT con
create_access_token() - Ritorna
TokenResponsecon access_token - Gestire errori: credenziali invalide (401)
Implementazione:
@router.post("/login", response_model=TokenResponse)
async def login(credentials: UserLogin, db: Session = Depends(get_db)):
# Trova utente
user = db.query(User).filter(User.email == credentials.email).first()
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials"
)
# Verifica password
if not verify_password(credentials.password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials"
)
# Genera JWT
access_token = create_access_token(data={"sub": str(user.id)})
return TokenResponse(
access_token=access_token,
expires_in=3600 # 1 ora
)
Test richiesti:
- Test login con credenziali valide successo
- Test login con email inesistente fallisce
- Test login con password sbagliata fallisce
- Test login utente disattivato fallisce
- Test JWT contiene user_id corretto
T20: Implementare Endpoint POST /api/auth/logout
Requisiti:
- Endpoint:
POST /api/auth/logout - Richiede autenticazione (JWT valido)
- In una implementazione JWT stateless, logout è gestito lato client
- Aggiungere token a blacklist (opzionale per MVP)
- Ritorna 200 con messaggio di successo
Implementazione:
@router.post("/logout")
async def logout(current_user: User = Depends(get_current_user)):
# In JWT stateless, il logout è gestito rimuovendo il token lato client
# Per implementazione con blacklist, aggiungere token a lista nera
return {"message": "Successfully logged out"}
Test richiesti:
- Test logout con token valido successo
- Test logout senza token fallisce (401)
- Test logout con token invalido fallisce (401)
T21: Implementare Dipendenza get_current_user
Requisiti:
- Creare
src/openrouter_monitor/dependencies/auth.py - Implementare
get_current_user()per FastAPI dependency injection - Estrae JWT da header Authorization (Bearer token)
- Verifica token con
verify_token()odecode_access_token() - Recupera utente dal DB per user_id nel token
- Verifica utente esista e sia attivo
- Gestire errori: token mancante, invalido, scaduto, utente non trovato
Implementazione:
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_db)
) -> User:
token = credentials.credentials
try:
payload = decode_access_token(token)
user_id = int(payload.get("sub"))
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token payload"
)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token"
)
user = db.query(User).filter(User.id == user_id).first()
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or inactive"
)
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
if not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user"
)
return current_user
Test richiesti:
- Test get_current_user con token valido ritorna utente
- Test get_current_user senza token fallisce
- Test get_current_user con token scaduto fallisce
- Test get_current_user con token invalido fallisce
- Test get_current_user utente non esiste fallisce
- Test get_current_user utente inattivo fallisce
T22: Scrivere Test per Auth Endpoints
Requisiti:
- Creare
tests/unit/routers/test_auth.py - Test integrazione per tutti gli endpoint auth
- Test con TestClient di FastAPI
- Mock database per test isolati
- Coverage >= 90%
Test richiesti:
-
Register Tests:
- POST /api/auth/register successo (201)
- POST /api/auth/register email duplicata (400)
- POST /api/auth/register password debole (422)
-
Login Tests:
- POST /api/auth/login successo (200 + token)
- POST /api/auth/login credenziali invalide (401)
- POST /api/auth/login utente inattivo (401)
-
Logout Tests:
- POST /api/auth/logout successo (200)
- POST /api/auth/logout senza token (401)
-
get_current_user Tests:
- Accesso protetto con token valido
- Accesso protetto senza token (401)
- Accesso protetto token scaduto (401)
🔄 WORKFLOW TDD OBBLIGATORIO
Per OGNI task (T17-T22):
┌─────────────────────────────────────────┐
│ 1. RED - Scrivi il test che fallisce │
│ • Test prima del codice │
│ • Pattern AAA (Arrange-Act-Assert) │
│ • Nomi descrittivi │
└─────────────────────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ 2. GREEN - Implementa codice minimo │
│ • Solo codice necessario per test │
│ • Nessun refactoring ancora │
└─────────────────────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ 3. REFACTOR - Migliora il codice │
│ • Pulisci duplicazioni │
│ • Migliora nomi variabili │
│ • Test rimangono verdi │
└─────────────────────────────────────────┘
📁 STRUTTURA FILE DA CREARE
src/openrouter_monitor/
├── schemas/
│ ├── __init__.py # Esporta tutti gli schemas
│ └── auth.py # T17 - Auth schemas
├── routers/
│ ├── __init__.py # Include auth router
│ └── auth.py # T18, T19, T20 - Auth endpoints
└── dependencies/
├── __init__.py
└── auth.py # T21 - get_current_user
tests/unit/
├── schemas/
│ ├── __init__.py
│ └── test_auth_schemas.py # T17 + T22
└── routers/
├── __init__.py
└── test_auth.py # T18, T19, T20, T21 + T22
🧪 REQUISITI TEST
Pattern AAA (Arrange-Act-Assert)
@pytest.mark.asyncio
async def test_register_new_user_returns_201_and_user_data():
# Arrange
user_data = {
"email": "test@example.com",
"password": "SecurePass123!",
"password_confirm": "SecurePass123!"
}
# Act
response = client.post("/api/auth/register", json=user_data)
# Assert
assert response.status_code == 201
assert response.json()["email"] == user_data["email"]
assert "id" in response.json()
Marker Pytest
@pytest.mark.unit # Logica pura
@pytest.mark.integration # Con database
@pytest.mark.asyncio # Funzioni async
@pytest.mark.auth # Test autenticazione
Fixtures Condivise
@pytest.fixture
def test_user(db_session):
"""Create test user in database."""
user = User(
email="test@example.com",
password_hash=hash_password("SecurePass123!")
)
db_session.add(user)
db_session.commit()
return user
@pytest.fixture
def auth_token(test_user):
"""Generate valid JWT for test user."""
return create_access_token(data={"sub": str(test_user.id)})
@pytest.fixture
def authorized_client(client, auth_token):
"""Client with authorization header."""
client.headers["Authorization"] = f"Bearer {auth_token}"
return client
🛡️ VINCOLI TECNICI
Pydantic Schemas Requirements
# Validazione password strength
def validate_password_strength(cls, v):
from openrouter_monitor.services.password import validate_password_strength
if not validate_password_strength(v):
raise ValueError(
'Password must be at least 12 characters with uppercase, '
'lowercase, digit, and special character'
)
return v
# Validazione passwords match
@root_validator
def passwords_match(cls, values):
if values.get('password') != values.get('password_confirm'):
raise ValueError('Passwords do not match')
return values
Router Requirements
from fastapi import APIRouter, Depends, HTTPException, status
router = APIRouter(prefix="/api/auth", tags=["auth"])
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(...):
...
@router.post("/login", response_model=TokenResponse)
async def login(...):
...
@router.post("/logout")
async def logout(...):
...
Dependency Requirements
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_db)
) -> User:
"""Extract and validate JWT, return current user."""
...
📊 AGGIORNAMENTO PROGRESS
Dopo ogni task completato, aggiorna:
/home/google/Sources/LucaSacchiNet/openrouter-watcher/export/progress.md
Esempio:
### 👤 Autenticazione Utenti (T17-T22)
- [x] T17: Pydantic schemas auth - Completato [timestamp]
- [x] T18: Endpoint POST /api/auth/register - Completato [timestamp]
- [ ] T19: Endpoint POST /api/auth/login - In progress
- [ ] T20: Endpoint POST /api/auth/logout
- [ ] T21: Dipendenza get_current_user
- [ ] T22: Test auth endpoints
**Progresso sezione:** 33% (2/6 task)
**Progresso totale:** 24% (18/74 task)
✅ CRITERI DI ACCETTAZIONE
- T17: Schemas auth completi con validazione
- T18: Endpoint /api/auth/register funzionante (201/400)
- T19: Endpoint /api/auth/login funzionante (200/401)
- T20: Endpoint /api/auth/logout funzionante
- T21: get_current_user dependency funzionante
- T22: Test completi per auth (coverage >= 90%)
- Tutti i test passano (
pytest tests/unit/routers/test_auth.py) - Nessuna password in plaintext nei log/errori
- 6 commit atomici (uno per task)
- progress.md aggiornato
🚀 COMANDO DI VERIFICA
Al termine, esegui:
cd /home/google/Sources/LucaSacchiNet/openrouter-watcher
pytest tests/unit/schemas/test_auth_schemas.py -v
pytest tests/unit/routers/test_auth.py -v --cov=src/openrouter_monitor/routers
# Verifica endpoint con curl
curl -X POST http://localhost:8000/api/auth/register \
-H "Content-Type: application/json" \
-d '{"email":"test@example.com","password":"SecurePass123!","password_confirm":"SecurePass123!"}'
curl -X POST http://localhost:8000/api/auth/login \
-H "Content-Type: application/json" \
-d '{"email":"test@example.com","password":"SecurePass123!"}'
🔒 CONSIDERAZIONI SICUREZZA
Do's ✅
- Usare
get_current_userper proteggere endpoint - Non loggare mai password in plaintext
- Ritornare errori generici per credenziali invalide
- Usare HTTPS in produzione
- Validare tutti gli input con Pydantic
Don'ts ❌
- MAI ritornare password hash nelle response
- MAI loggare token JWT completi
- MAI usare GET per operazioni che modificano dati
- MAI ignorare eccezioni di autenticazione
📝 NOTE
- Usa SEMPRE path assoluti:
/home/google/Sources/LucaSacchiNet/openrouter-watcher/ - Segui le convenzioni in
.opencode/agents/tdd-developer.md - Task devono essere verificabili in < 2 ore ciascuno
- Documenta bug complessi in
/docs/bug_ledger.md - Usa conventional commits:
feat(schemas): T17 add Pydantic auth schemasfeat(auth): T18 implement user registration endpointfeat(auth): T19 implement user login endpointfeat(auth): T20 implement user logout endpointfeat(deps): T21 implement get_current_user dependencytest(auth): T22 add comprehensive auth endpoint tests
AGENTE: @tdd-developer
INIZIA CON: T17 - Pydantic schemas