security(CASA-28): API Key security enhancements

Author: orhanrauf

backend/airweave/api/deps.py

@@ -59,8 +59,33 @@ async def _authenticate_auth0_user(
     return user_context, AuthMethod.AUTH0, {"auth0_id": auth0_user.id}
 
 
+def _extract_client_ip(request: Request) -> str:
+    """Extract client IP from request headers.
+
+    Checks X-Forwarded-For header first (for proxied requests),
+    then falls back to direct client IP.
+
+    Args:
+    ----
+        request (Request): FastAPI request object
+
+    Returns:
+    -------
+        str: Client IP address or "unknown" if not available
+
+    """
+    # Check X-Forwarded-For first (for proxied requests)
+    forwarded_for = request.headers.get("X-Forwarded-For")
+    if forwarded_for:
+        # X-Forwarded-For can be a comma-separated list, take the first one (original client)
+        return forwarded_for.split(",")[0].strip()
+
+    # Fallback to direct client IP
+    return request.client.host if request.client else "unknown"
+
+
 async def _authenticate_api_key(
-    db: AsyncSession, api_key: str
+    db: AsyncSession, api_key: str, request: Request
 ) -> Tuple[None, AuthMethod, dict, str]:
     """Authenticate API key and return organization ID.
 
@@ -87,6 +112,14 @@ async def _authenticate_api_key(
         api_key_obj = await crud.api_key.get_by_key(db, key=api_key)
         org_id = api_key_obj.organization_id
 
+        # Log API key usage with structured dimensions (flows to Azure LAW)
+        client_ip = _extract_client_ip(request)
+        audit_logger = logger.with_context(event_type="api_key_usage")
+        audit_logger.info(
+            f"API key usage: key={api_key_obj.id} org={org_id} ip={client_ip} "
+            f"endpoint={request.url.path} created_by={api_key_obj.created_by_email}"
+        )
+
         # Cache the mapping for next time
         await context_cache.set_api_key_org_id(api_key, org_id)
 
@@ -328,7 +361,7 @@ async def get_context(
         user_context, auth_method, auth_metadata = await _get_or_fetch_user_context(db, auth0_user)
     elif x_api_key:
         user_context, auth_method, auth_metadata, api_key_org_id = await _authenticate_api_key(
-            db, x_api_key
+            db, x_api_key, request
         )
 
     if not auth_method:

backend/airweave/api/v1/endpoints/api_keys.py

@@ -10,6 +10,7 @@
 from airweave.api.context import ApiContext
 from airweave.api.router import TrailingSlashRouter
 from airweave.core import credentials
+from airweave.core.datetime_utils import utc_now_naive
 
 router = TrailingSlashRouter()
 
@@ -18,7 +19,7 @@
 async def create_api_key(
     *,
     db: AsyncSession = Depends(deps.get_db),
-    api_key_in: schemas.APIKeyCreate = Body({}),  # Default to empty dict if not provided
+    api_key_in: schemas.APIKeyCreate = Body(default_factory=lambda: schemas.APIKeyCreate()),
     ctx: ApiContext = Depends(deps.get_context),
 ) -> schemas.APIKey:
     """Create a new API key for the current user.
@@ -39,13 +40,22 @@ async def create_api_key(
     """
     api_key_obj = await crud.api_key.create(db=db, obj_in=api_key_in, ctx=ctx)
 
+    # Audit log: API key creation (flows to Azure LAW)
+    expiration_days = (api_key_obj.expiration_date - api_key_obj.created_at).days
+    audit_logger = ctx.logger.with_context(event_type="api_key_created")
+    audit_logger.info(
+        f"API key created: {api_key_obj.id} by {api_key_obj.created_by_email} "
+        f"for org {ctx.organization.id}, expires in {expiration_days} days "
+        f"({api_key_obj.expiration_date.isoformat()})"
+    )
+
     # Decrypt the key for the response
     decrypted_data = credentials.decrypt(api_key_obj.encrypted_key)
     decrypted_key = decrypted_data["key"]
 
     api_key_data = {
         "id": api_key_obj.id,
-        "organization": ctx.organization.id,  # Use the user's organization_id
+        "organization_id": ctx.organization.id,
         "created_at": api_key_obj.created_at,
         "modified_at": api_key_obj.modified_at,
         "last_used_date": None,  # New key has no last used date
@@ -88,7 +98,7 @@ async def read_api_key(
 
     api_key_data = {
         "id": api_key.id,
-        "organization": ctx.organization.id,
+        "organization_id": ctx.organization.id,
         "created_at": api_key.created_at,
         "modified_at": api_key.modified_at,
         "last_used_date": api_key.last_used_date if hasattr(api_key, "last_used_date") else None,
@@ -132,7 +142,7 @@ async def read_api_keys(
 
         api_key_data = {
             "id": api_key.id,
-            "organization": ctx.organization.id,
+            "organization_id": ctx.organization.id,
             "created_at": api_key.created_at,
             "modified_at": api_key.modified_at,
             "last_used_date": (
@@ -148,6 +158,63 @@ async def read_api_keys(
     return result
 
 
+@router.post("/{id}/rotate", response_model=schemas.APIKey)
+async def rotate_api_key(
+    *,
+    db: AsyncSession = Depends(deps.get_db),
+    id: UUID,
+    ctx: ApiContext = Depends(deps.get_context),
+) -> schemas.APIKey:
+    """Rotate an API key by creating a new one.
+
+    This endpoint creates a new API key with a fresh 90-day expiration.
+    The old key remains active until its original expiration date.
+    Users can manage multiple keys or delete the old one manually if desired.
+
+    Args:
+    ----
+        db (AsyncSession): The database session.
+        id (UUID): The ID of the API key to rotate.
+        ctx (ApiContext): The current authentication context.
+
+    Returns:
+    -------
+        schemas.APIKey: The newly created API key with decrypted key value.
+
+    Raises:
+    ------
+        HTTPException: If the API key is not found or user doesn't have access.
+
+    """
+    # Verify old key exists and user has access
+    old_key = await crud.api_key.get(db=db, id=id, ctx=ctx)
+    old_key_schema = schemas.APIKey.model_validate(old_key, from_attributes=True)
+
+    # Create new key with default 90-day expiration
+    new_key_obj = await crud.api_key.create(
+        db=db,
+        obj_in=schemas.APIKeyCreate(),  # Uses default 90 days
+        ctx=ctx,
+    )
+
+    # Decrypt the new key for the response
+    decrypted_data = credentials.decrypt(new_key_obj.encrypted_key)
+    decrypted_key = decrypted_data["key"]
+
+    new_key_schema = schemas.APIKey.model_validate(new_key_obj, from_attributes=True)
+    new_key_schema.decrypted_key = decrypted_key
+
+    # Audit log: API key rotation (flows to Azure LAW)
+    audit_logger = ctx.logger.with_context(event_type="api_key_rotated")
+    audit_logger.info(
+        f"API key rotated: old={old_key_schema.id}, new={new_key_schema.id} "
+        f"by {new_key_schema.created_by_email} for org {ctx.organization.id}, "
+        f"new key expires {new_key_schema.expiration_date.isoformat()}"
+    )
+
+    return new_key_schema
+
+
 @router.delete("/", response_model=schemas.APIKey)
 async def delete_api_key(
     *,
@@ -181,7 +248,7 @@ async def delete_api_key(
     # Create a copy of the data before deletion
     api_key_data = {
         "id": api_key.id,
-        "organization": ctx.organization.id,
+        "organization_id": ctx.organization.id,
         "created_at": api_key.created_at,
         "modified_at": api_key.modified_at,
         "last_used_date": api_key.last_used_date if hasattr(api_key, "last_used_date") else None,
@@ -191,6 +258,14 @@ async def delete_api_key(
         "decrypted_key": decrypted_key,
     }
 
+    # Audit log: API key deletion (flows to Azure LAW)
+    was_expired = api_key.expiration_date < utc_now_naive()
+    audit_logger = ctx.logger.with_context(event_type="api_key_deleted")
+    audit_logger.info(
+        f"API key deleted: {api_key.id} by {ctx.tracking_email} for org {ctx.organization.id} "
+        f"(was_expired={was_expired})"
+    )
+
     # Now delete the API key
     await crud.api_key.remove(db=db, id=id, ctx=ctx)
 

backend/airweave/api/v1/endpoints/users.py

@@ -17,11 +17,11 @@
 from airweave.api.auth import auth0
 from airweave.api.context import ApiContext
 from airweave.api.router import TrailingSlashRouter
-from airweave.core.email_service import send_welcome_email
 from airweave.core.exceptions import NotFoundException
 from airweave.core.logging import logger
 from airweave.core.shared_models import AuthMethod
 from airweave.db.unit_of_work import UnitOfWork
+from airweave.email.services import send_welcome_email
 from airweave.schemas import OrganizationWithRole, User
 
 router = TrailingSlashRouter()

backend/airweave/crud/crud_api_key.py

@@ -1,7 +1,7 @@
 """CRUD operations for the APIKey model."""
 
 import secrets
-from datetime import timedelta
+from datetime import datetime, timedelta
 from typing import Optional
 from uuid import UUID
 
@@ -45,9 +45,9 @@ async def create(
         key = secrets.token_urlsafe(32)
         encrypted_key = credentials.encrypt({"key": key})
 
-        expiration_date = obj_in.expiration_date or (
-            utc_now_naive() + timedelta(days=180)  # Default to 180 days
-        )
+        # Calculate expiration date from days (defaults to 90)
+        expiration_days = obj_in.expiration_days if obj_in.expiration_days is not None else 90
+        expiration_date = utc_now_naive() + timedelta(days=expiration_days)
 
         # Create a dictionary with the data instead of using the schema
         api_key_data = {
@@ -143,5 +143,36 @@ async def get_by_key(self, db: AsyncSession, *, key: str) -> Optional[APIKey]:
 
         raise NotFoundException("API key not found")
 
+    async def get_keys_expiring_in_range(
+        self,
+        db: AsyncSession,
+        start_date: datetime,
+        end_date: datetime,
+    ) -> list[APIKey]:
+        """Get API keys expiring within a date range.
+
+        Args:
+        ----
+            db (AsyncSession): The database session.
+            start_date (datetime): Start of the date range (inclusive).
+            end_date (datetime): End of the date range (exclusive).
+
+        Returns:
+        -------
+            list[APIKey]: List of API keys expiring in the range.
+
+        """
+        from sqlalchemy import and_, select
+
+        query = select(self.model).where(
+            and_(
+                self.model.expiration_date >= start_date,
+                self.model.expiration_date < end_date,
+            )
+        )
+
+        result = await db.execute(query)
+        return list(result.scalars().all())
+
 
 api_key = CRUDAPIKey(APIKey)

backend/airweave/email/__init__.py

@@ -0,0 +1,10 @@
+"""Email module for Airweave."""
+
+from airweave.email.services import send_email_via_resend, send_welcome_email
+from airweave.email.templates import get_api_key_expiration_email
+
+__all__ = [
+    "send_email_via_resend",
+    "send_welcome_email",
+    "get_api_key_expiration_email",
+]

backend/airweave/core/email_service.py renamed to backend/airweave/email/services.py

@@ -1,15 +1,91 @@
-"""Simple email service for sending welcome emails via Resend."""
+"""Email service for sending emails via Resend."""
 
 import asyncio
 import random
 from datetime import datetime, timedelta, timezone
+from typing import Optional
 
 import resend
 
 from airweave.core.config import settings
 from airweave.core.logging import logger
 
 
+def _send_email_via_resend_sync(
+    to_email: str,
+    subject: str,
+    html_body: str,
+    from_email: Optional[str] = None,
+    scheduled_at: Optional[str] = None,
+) -> None:
+    """Synchronous email sending function via Resend to be run in a thread pool.
+
+    Args:
+    ----
+        to_email (str): Recipient email address
+        subject (str): Email subject line
+        html_body (str): HTML email body
+        from_email (Optional[str]): Sender email (defaults to settings.RESEND_FROM_EMAIL)
+        scheduled_at (Optional[str]): ISO 8601 timestamp for scheduled delivery
+
+    """
+    resend.api_key = settings.RESEND_API_KEY
+
+    email_data = {
+        "from": from_email or settings.RESEND_FROM_EMAIL,
+        "to": [to_email],
+        "subject": subject,
+        "html": html_body,
+    }
+
+    if scheduled_at:
+        email_data["scheduled_at"] = scheduled_at
+
+    resend.Emails.send(email_data)
+
+
+async def send_email_via_resend(
+    to_email: str,
+    subject: str,
+    html_body: str,
+    from_email: Optional[str] = None,
+    scheduled_at: Optional[str] = None,
+) -> bool:
+    """Send an email via Resend asynchronously.
+
+    Args:
+    ----
+        to_email (str): Recipient email address
+        subject (str): Email subject line
+        html_body (str): HTML email body
+        from_email (Optional[str]): Sender email (defaults to settings.RESEND_FROM_EMAIL)
+        scheduled_at (Optional[str]): ISO 8601 timestamp for scheduled delivery
+
+    Returns:
+    -------
+        bool: True if email was sent successfully, False otherwise
+
+    """
+    if not settings.RESEND_API_KEY or not settings.RESEND_FROM_EMAIL:
+        logger.debug("RESEND_API_KEY or RESEND_FROM_EMAIL not configured - skipping email")
+        return False
+
+    try:
+        await asyncio.to_thread(
+            _send_email_via_resend_sync,
+            to_email,
+            subject,
+            html_body,
+            from_email,
+            scheduled_at,
+        )
+        logger.info(f"Email sent to {to_email}: {subject}")
+        return True
+    except Exception as e:
+        logger.error(f"Failed to send email to {to_email}: {e}")
+        return False
+
+
 def _send_welcome_email_sync(to_email: str, user_name: str) -> None:
     """Synchronous email sending function to be run in a thread pool."""
     resend.api_key = settings.RESEND_API_KEY