Skip to content
Module 4 · API & Business LogicIntermediate2 min read · 389 words

Lesson 11: Authentication

"Security is not optional."

Current State

⚠️ No authentication implemented - All endpoints are open

Why Authentication Needed

  • Prevent unauthorized access
  • Track who does what
  • Rate limiting per user
  • Data isolation

JWT Implementation Guide

Step 1: Add Dependencies

python
pip install python-jose[cryptography] passlib[bcrypt]

Step 2: Create User Model

python
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True)
    hashed_password = Column(String)

Step 3: Password Hashing

python
from passlib.context import CryptContext
 
pwd_context = CryptContext(schemes=["bcrypt"])
 
def hash_password(password: str) -> str:
    return pwd_context.hash(password)
 
def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

Step 4: JWT Tokens

python
from jose import JWTError, jwt
from datetime import datetime, timedelta
 
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
 
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(hours=24)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 
def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

Step 5: Login Endpoint

python
@router.post("/login")
def login(email: str, password: str, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == email).first()
    if not user or not verify_password(password, user.hashed_password):
        raise HTTPException(401, "Invalid credentials")
    
    token = create_access_token({"user_id": user.id})
    return {"access_token": token, "token_type": "bearer"}

Step 6: Protect Endpoints

python
def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    payload = verify_token(token)
    if not payload:
        raise HTTPException(401, "Invalid token")
    
    user = db.query(User).filter(User.id == payload["user_id"]).first()
    return user
 
@router.get("/agents/")
def get_agents(
    user = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    # Only return user's agents
    return db.query(Agent).filter(Agent.user_id == user.id).all()

OAuth2 Alternative

For social login (Google, GitHub):

python
from authlib.integrations.starlette_client import OAuth
 
oauth = OAuth()
oauth.register(
    name='google',
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

Security Best Practices

✅ Never store passwords in plain text ✅ Use HTTPS in production ✅ Set short token expiration ✅ Implement refresh tokens ✅ Rate limit login attempts ✅ Log authentication events

Key Takeaways

🎯 VeyraOps AI currently has no auth (development only) 🎯 JWT is industry standard for API authentication 🎯 Hash passwords with bcrypt 🎯 Protect endpoints with dependencies 🎯 OAuth2 for social login

👉 Lesson 12: Workflows →

Module 4 · API & Business Logic

PreviousLesson 10 - Dependency InjectionNextLesson 12 - Multi-Agent Workflows