OWASP API सुरक्षा के शीर्ष 10 जोखिमों का शमन

import os
from datetime import datetime, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

SECRET_KEY = os.environ["JWT_SECRET_KEY"]  # never hardcode
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


def get_current_user_id(token: str = Depends(oauth2_scheme)) -> int:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
        return int(user_id)
    except JWTError:
        raise credentials_exception

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import get_current_user_id
from app.db import get_db
from app.models import Order

router = APIRouter()


@router.get("/orders/{order_id}")
async def get_order(
    order_id: int,
    current_user_id: int = Depends(get_current_user_id),
    db: AsyncSession = Depends(get_db),
):
    order = await db.get(Order, order_id)
    if order is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Order not found")

    # BOLA fix: verify the caller owns this resource
    if order.user_id != current_user_id:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")

    return order

from fastapi import HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession


async def get_owned_resource(model_class, resource_id: int, user_id: int, db: AsyncSession):
    """
    Fetch a resource and assert ownership in one call.
    Raises 404 if not found, 403 if the caller does not own it.
    """
    resource = await db.get(model_class, resource_id)
    if resource is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"{model_class.__name__} not found",
        )
    if resource.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not have permission to access this resource",
        )
    return resource

from pydantic import BaseModel, EmailStr
from typing import Optional


# What the client is ALLOWED to send when updating a profile
class UserUpdateInput(BaseModel):
    display_name: Optional[str] = None
    bio: Optional[str] = None
    email: Optional[EmailStr] = None
    # NOTE: is_admin, balance, role, hashed_password are NOT here


# What we return to the client (read-only fields visible but not settable)
class UserPublicOutput(BaseModel):
    id: int
    display_name: str
    email: EmailStr
    is_admin: bool

    model_config = {"from_attributes": True}

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import get_current_user_id
from app.db import get_db
from app.models import User
from app.schemas import UserUpdateInput, UserPublicOutput

router = APIRouter()


@router.patch("/users/me", response_model=UserPublicOutput)
async def update_profile(
    payload: UserUpdateInput,
    current_user_id: int = Depends(get_current_user_id),
    db: AsyncSession = Depends(get_db),
):
    user = await db.get(User, current_user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    # Only apply fields the client actually provided
    update_data = payload.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(user, field, value)

    await db.commit()
    await db.refresh(user)
    return user

from pydantic import BaseModel, EmailStr, ConfigDict
from typing import Optional


class StrictUserUpdateInput(BaseModel):
    model_config = ConfigDict(extra="forbid")  # reject unknown fields

    display_name: Optional[str] = None
    bio: Optional[str] = None
    email: Optional[EmailStr] = None


# Demonstration (runs standalone)
if __name__ == "__main__":
    import json
    from pydantic import ValidationError

    # Valid input
    valid = StrictUserUpdateInput(display_name="Alice")
    print("Valid:", valid.model_dump(exclude_unset=True))

    # Attacker tries to escalate privileges
    try:
        evil = StrictUserUpdateInput(display_name="Alice", is_admin=True)
    except ValidationError as e:
        errors = json.loads(e.json())
        print("Blocked:", errors[0]["type"], "-", errors[0]["loc"])

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, ConfigDict
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import get_current_user_id
from app.db import get_db
from app.models import Document

router = APIRouter()


class DocumentUpdateInput(BaseModel):
    model_config = ConfigDict(extra="forbid")
    title: Optional[str] = None
    content: Optional[str] = None


@router.patch("/documents/{doc_id}")
async def update_document(
    doc_id: int,
    payload: DocumentUpdateInput,                          # Layer 3: mass assignment guard
    current_user_id: int = Depends(get_current_user_id),  # Layer 1: auth
    db: AsyncSession = Depends(get_db),
):
    doc = await db.get(Document, doc_id)
    if doc is None:
        raise HTTPException(status_code=404, detail="Document not found")

    if doc.owner_id != current_user_id:                   # Layer 2: BOLA guard
        raise HTTPException(status_code=403, detail="Access denied")

    for field, value in payload.model_dump(exclude_unset=True).items():
        setattr(doc, field, value)

    await db.commit()
    await db.refresh(doc)
    return doc