Merge pull request #28 from codewithme-py/feat/security-fixes

feat: implement login/signup rate limiting, add background image sanitization tasks, and enhance order detail access auditing.

Author: codewithme-py

.env

@@ -34,6 +34,10 @@ REFRESH_TOKEN_EXPIRE_DAYS=7
 RATE_LIMIT_USER_RPS=10
 RATE_LIMIT_GLOBAL_RPS=1000
 RATE_LIMIT_TTL_SECONDS=1
+LOGIN_RATE_LIMIT_ATTEMPTS=7
+LOGIN_RATE_LIMIT_TTL=60
+SIGNUP_RATE_LIMIT_ATTEMPTS=3
+SIGNUP_RATE_LIMIT_TTL=3600
 IDEMPOTENT_KEY_LIFETIME_SEC=86400
 RESERVE_TIMEOUT_MINUTES=15
 #db_engine_layer

app/core/admin/admin.py

@@ -82,10 +82,10 @@ def docs_link_formatter(model: Any, name: Any) -> Any:
             return 'No docs'
         links = []
         if isinstance(docs, dict):
-            for doc_type, s3_key in docs.items():
+            for doc_type in docs.keys():
                 links.append(
-                    f'<a href="/api/v1/media/view?key={s3_key}" '
-                    f'target="_blank">{doc_type}</a>'
+                    f'<a href="/api/v1/media/view/verification_doc/{model.id}'
+                    f'?doc_key={doc_type}" target="_blank">{doc_type}</a>'
                 )
         return Markup(', '.join(links))
 

app/core/audit_log/service.py

@@ -7,6 +7,11 @@
 
 from app.core.audit_log.models import AuditLog
 
+SENSITIVE_FIELDS = {
+    'shipping_address',
+    'email',
+}
+
 
 class AuditLogService:
     async def log_event(
@@ -41,20 +46,24 @@ def get_diff(
         old_model: BaseModel | None,
         new_model: BaseModel | None,
     ) -> dict[str, Any]:
+        diff: dict[str, Any] = {}
         if old_model is None and new_model is not None:
             return {k: [None, v] for k, v in new_model.model_dump(mode='json').items()}
         if old_model is not None and new_model is None:
             return {k: [v, None] for k, v in old_model.model_dump(mode='json').items()}
         if old_model is not None and new_model is not None:
             old_data = old_model.model_dump(mode='json')
             new_data = new_model.model_dump(mode='json')
-            diff = {}
             for key, value in new_data.items():
                 old_val = old_data.get(key)
                 if value != old_val:
-                    diff[key] = [old_val, value]
+                    if key in SENSITIVE_FIELDS:
+                        masked = '[SENSITIVE_DATA_HIDDEN]'
+                        diff[key] = [masked, masked]
+                    else:
+                        diff[key] = [old_val, value]
             return diff
-        return {}
+        return diff
 
     async def log_object_change(
         self,
@@ -79,5 +88,22 @@ async def log_object_change(
                 extra_data=extra_data,
             )
 
+    async def log_pii_access(
+        self,
+        session: AsyncSession,
+        actor_id: UUID,
+        target_id: UUID,
+        target_type: str,
+        reason: str | None = None,
+    ) -> None:
+        await self.log_event(
+            session=session,
+            actor_id=actor_id,
+            target_id=target_id,
+            target_type=target_type,
+            action='pii_access',
+            changes={'pii_accessed': True, 'reason': reason},
+        )
+
 
 audit_log_service = AuditLogService()

app/core/config.py

@@ -28,6 +28,10 @@ class Settings(BaseSettings):
     rate_limit_user_rps: int = Field(alias='RATE_LIMIT_USER_RPS')
     rate_limit_global_rps: int = Field(alias='RATE_LIMIT_GLOBAL_RPS')
     rate_limit_ttl_seconds: int = Field(alias='RATE_LIMIT_TTL_SECONDS')
+    login_rate_limit_attempts: int = Field(alias='LOGIN_RATE_LIMIT_ATTEMPTS')
+    login_rate_limit_ttl: int = Field(alias='LOGIN_RATE_LIMIT_TTL')
+    signup_rate_limit_attempts: int = Field(alias='SIGNUP_RATE_LIMIT_ATTEMPTS')
+    signup_rate_limit_ttl: int = Field(alias='SIGNUP_RATE_LIMIT_TTL')
     idempotent_key_lifetime_sec: int = Field(alias='IDEMPOTENT_KEY_LIFETIME_SEC')
     reserve_timeout_minutes: int = Field(alias='RESERVE_TIMEOUT_MINUTES')
     presigned_url_expire_seconds: int = Field(alias='PRESIGNED_URL_EXPIRE_SECONDS')

app/core/s3.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
 from typing import Any
 
 import aioboto3  # type: ignore
@@ -12,6 +13,7 @@
 session = aioboto3.Session()
 
 
+@asynccontextmanager
 async def get_s3_client() -> AsyncGenerator[Any, None]:
     async with session.client(
         's3',

app/main.py

@@ -3,6 +3,8 @@
 from contextlib import asynccontextmanager
 
 import structlog
+from arq import create_pool
+from arq.connections import RedisSettings
 from fastapi import FastAPI, Request, Response
 from fastapi.responses import ORJSONResponse
 from prometheus_fastapi_instrumentator import Instrumentator
@@ -36,11 +38,14 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     client = Redis.from_url(settings.redis_url, decode_responses=True, encoding='utf-8')
     app.state.redis = client
     app.state.rate_limit_script = client.register_script(RATE_LIMIT_LUA_SCRIPT)
+    app.state.arq_redis = await create_pool(RedisSettings.from_dsn(settings.redis_url))
     await init_s3_bucket()
     try:
         logger.info('redis connected')
+        logger.info('arq pool created')
         yield
     finally:
+        await app.state.arq_redis.close()
         await client.aclose()
         logger.info('redis disconnected')
 

app/services/inventory/routes.py

@@ -3,10 +3,10 @@
 
 from fastapi import APIRouter, Depends, Header, Query, Request, status
 
+from app.core.config import settings
 from app.core.database import SessionDep
 from app.core.security import RoleChecker, UserRole
 from app.services.inventory.models import ProductStatus
-from app.services.inventory.rate_limit import check_rate_limit
 from app.services.inventory.schemas import (
     ProductCreate,
     ProductRead,
@@ -18,6 +18,7 @@
 from app.services.user.models import User
 from app.shared.decorators import idempotent
 from app.shared.deps import get_current_user
+from app.shared.rate_limit import check_rate_limit
 
 from .deps import (
     get_inventory_admin_service,
@@ -153,8 +154,11 @@ async def reservation_data(
 ) -> ReservationResponse:
     await check_rate_limit(
         rate_limit_script=request.app.state.rate_limit_script,
-        user_id=str(current_user.id),
-        item_id=str(reservation_data.product_id),
+        keys=[
+            f'rate_limit:user:{current_user.id}',
+            f'rate_limit:item:{reservation_data.product_id}',
+        ],
+        limits=[settings.rate_limit_user_rps, settings.rate_limit_global_rps],
     )
     result = await service.reserve_items(
         session=session,

app/services/media/routes.py

@@ -1,10 +1,12 @@
+from http import HTTPStatus
 from typing import Any
 from uuid import UUID
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import RedirectResponse
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.audit_log.service import audit_log_service
 from app.core.database import get_session
 from app.core.s3 import get_s3_client
 from app.services.media.schemas import (
@@ -15,6 +17,7 @@
 from app.services.media.service import (
     generate_presigned_get_url,
     generate_upload_url,
+    get_secure_file_path,
     handle_minio_webhook,
 )
 from app.services.user.models import User, UserRole
@@ -36,23 +39,38 @@ async def create_upload_url(
 
 @router_v1.post('/webhook/minio')
 async def minio_webhook(
+    request: Request,
     event: MinioWebhookEvent,
     session: AsyncSession = Depends(get_session),
 ) -> dict[str, str]:
-    await handle_minio_webhook(session, event)
+    await handle_minio_webhook(session, event, arq_redis=request.app.state.arq_redis)
     return {'status': 'ok'}
 
 
-@router_v1.get('/view', response_class=RedirectResponse)
+@router_v1.get('/view/{target_type}/{target_id}', response_class=RedirectResponse)
 async def view_private_file(
-    key: str,
+    target_type: str,
+    target_id: UUID,
+    doc_key: str | None = None,
+    session: AsyncSession = Depends(get_session),
     s3_client: Any = Depends(get_s3_client),
     current_user: User = Depends(get_current_user),
 ) -> RedirectResponse:
     if current_user.role not in (UserRole.ADMIN, UserRole.MODERATOR):
-        from fastapi import HTTPException
-
-        raise HTTPException(status_code=403, detail='Not authorized to view this file')
-
-    url = await generate_presigned_get_url(s3_client, key)
-    return RedirectResponse(url=url, status_code=307)
+        raise HTTPException(
+            status_code=HTTPStatus.FORBIDDEN, detail='Not authorized to view this file'
+        )
+    file_path = await get_secure_file_path(session, target_type, target_id, doc_key)
+    if not file_path:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail='File not found')
+    if target_type == 'verification_doc':
+        await audit_log_service.log_pii_access(
+            session=session,
+            actor_id=current_user.id,
+            target_id=target_id,
+            target_type='verification_request',
+            reason=f'viewing_doc_{doc_key}' if doc_key else 'viewing_verification_doc',
+        )
+        await session.commit()
+    url = await generate_presigned_get_url(s3_client, file_path)
+    return RedirectResponse(url=url, status_code=HTTPStatus.TEMPORARY_REDIRECT)

app/services/media/service.py

@@ -14,6 +14,7 @@
     ImageUploadResponse,
     MinioWebhookEvent,
 )
+from app.services.user.models import VerificationRequest
 
 logger = structlog.get_logger(__name__)
 
@@ -34,6 +35,39 @@ async def generate_presigned_get_url(
     )
 
 
+async def get_secure_file_path(
+    session: AsyncSession,
+    target_type: str,
+    target_id: UUID,
+    doc_key: str | None = None,
+) -> str | None:
+    """Resolves a secure S3 path from a database resource."""
+    if target_type == 'verification_doc':
+        result_v = await session.execute(
+            select(VerificationRequest).where(VerificationRequest.id == target_id)
+        )
+        v_req = result_v.scalar_one_or_none()
+        if v_req and v_req.docs_url and doc_key:
+            return str(v_req.docs_url.get(doc_key))
+    elif target_type == 'product_image':
+        result_i = await session.execute(
+            select(ProductImage).where(ProductImage.id == target_id)
+        )
+        img = result_i.scalar_one_or_none()
+        if img:
+            return str(img.file_path)
+    return None
+
+
+async def sanitize_image_metadata(
+    s3_client: Any,
+    bucket: str,
+    key: str,
+) -> None:
+    """OBSOLETE: Moved to tasks.py"""
+    pass
+
+
 async def generate_upload_url(
     session: AsyncSession,
     s3_client: Any,
@@ -74,6 +108,7 @@ async def generate_upload_url(
 async def handle_minio_webhook(
     session: AsyncSession,
     event: MinioWebhookEvent,
+    arq_redis: Any = None,
 ) -> None:
     if not event.records:
         return
@@ -88,7 +123,16 @@ async def handle_minio_webhook(
         )
         image = result.scalar_one_or_none()
         if image is not None and image.status == ImageStatus.PENDING:
-            image.status = ImageStatus.ACTIVE
-            await session.commit()
+            # Point 3: Enqueue background sanitization task
+            if arq_redis:
+                await arq_redis.enqueue_job(
+                    'sanitize_and_activate_image_task',
+                    image_id=image.id,
+                    bucket=settings.minio_bucket_name,
+                    object_key=object_key,
+                )
+                logger.info('enqueued image sanitization task', key=object_key)
+            else:
+                logger.warning('arq_redis pool not provided to webhook handler')
         else:
             logger.warning('image not found or not in pending status')

app/services/media/tasks.py

@@ -0,0 +1,45 @@
+import io
+from uuid import UUID
+
+from PIL import Image
+from sqlalchemy import select
+
+from app.core.s3 import get_s3_client
+from app.services.media.models import ImageStatus, ProductImage
+
+
+async def sanitize_and_activate_image_task(
+    ctx: dict,
+    image_id: UUID,
+    bucket: str,
+    object_key: str,
+) -> None:
+    """Background task to strip EXIF and activate an image."""
+    session_maker = ctx['session_maker']
+
+    # 1. Sanitize the image
+    async with get_s3_client() as s3_client:
+        response = await s3_client.get_object(Bucket=bucket, Key=object_key)
+        image_data = await response['Body'].read()
+
+        with Image.open(io.BytesIO(image_data)) as img:
+            output = io.BytesIO()
+            img.save(output, format=img.format)
+            output.seek(0)
+
+            await s3_client.put_object(
+                Bucket=bucket,
+                Key=object_key,
+                Body=output,
+                ContentType=response.get('ContentType', 'image/jpeg'),
+            )
+
+    # 2. Update status in database
+    async with session_maker() as session:
+        result = await session.execute(
+            select(ProductImage).where(ProductImage.id == image_id)
+        )
+        image = result.scalar_one_or_none()
+        if image:
+            image.status = ImageStatus.ACTIVE
+            await session.commit()