Spaces:
Sleeping
Sleeping
File size: 5,709 Bytes
30384dd d35d068 30384dd d35d068 30384dd d35d068 30384dd 5c6567b 92bf88b 5c6567b 92bf88b 5c6567b 92bf88b 5c6567b 92bf88b 5c6567b 92bf88b 5c6567b 92bf88b 5c6567b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | from datetime import datetime, timezone
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.core.security import verify_api_key
from app.db.session import get_db
from app.models.api_key import ApiKey
from app.models.user import User
router = APIRouter(tags=["auth"])
security = HTTPBearer(auto_error=False)
class UserResponse(BaseModel):
"""User response model."""
id: int = Field(..., description="User ID")
email: str = Field(..., description="User email")
created_at: str = Field(..., description="ISO 8601 timestamp in UTC")
class Config:
json_schema_extra = {
"example": {
"id": 1,
"email": "user@example.com",
"created_at": "2025-12-31T00:00:00Z"
}
}
def get_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security),
db: Session = Depends(get_db)
) -> User:
"""
Authenticate user via API key from Authorization header.
Raises 401 if authentication fails.
"""
if credentials is None:
raise HTTPException(
status_code=401,
detail="Authorization header missing"
)
if db is None:
raise HTTPException(
status_code=503,
detail="Database not available"
)
# Extract API key from Bearer token
api_key = credentials.credentials
# Find API key in database
api_keys = db.query(ApiKey).all()
matching_key: Optional[ApiKey] = None
for key in api_keys:
if verify_api_key(api_key, key.key_hash):
# Check if key is expired
if key.expires_at and key.expires_at < datetime.now(timezone.utc):
continue
matching_key = key
# Update last_used_at
key.last_used_at = datetime.now(timezone.utc)
db.commit()
break
if matching_key is None:
raise HTTPException(
status_code=401,
detail="Invalid or expired API key"
)
# Get user
user = db.query(User).filter(User.id == matching_key.user_id).first()
if user is None:
raise HTTPException(
status_code=404,
detail="User not found"
)
return user
@router.get("/me", response_model=UserResponse)
def get_me(current_user: User = Depends(get_current_user)) -> UserResponse:
"""
Get current authenticated user information.
Requires valid API key in Authorization header.
Returns 401 if authentication fails.
"""
return UserResponse(
id=current_user.id,
email=current_user.email,
created_at=current_user.created_at.isoformat() + "Z"
)
@router.get("/admin/api-key")
def get_admin_api_key(db: Session = Depends(get_db)):
"""
Get the seeded admin API key.
This endpoint tries to read from file first, if not found,
it will generate a new one from the database (if seed ran successfully).
"""
import os
from pathlib import Path
# Try multiple file paths
possible_paths = [
Path("admin_api_key.txt"),
Path("/tmp/admin_api_key.txt"),
Path("/app/admin_api_key.txt"),
Path("./admin_api_key.txt"),
]
key_file_path = None
for path in possible_paths:
if path.exists():
key_file_path = path
break
# If file exists, read from it
if key_file_path:
try:
with open(key_file_path, "r") as f:
content = f.read().strip()
lines = content.split("\n")
api_key = None
email = None
for line in lines:
if line.startswith("API Key:"):
api_key = line.replace("API Key:", "").strip()
elif line.startswith("Email:"):
email = line.replace("Email:", "").strip()
if api_key:
return {
"email": email or "admin@forgeflow.local",
"api_key": api_key,
"source": "file",
"message": "⚠️ IMPORTANT: Copy this API key. It will not be shown again after you use it!"
}
except Exception as e:
pass # Fall through to DB check
# If file not found, try to get from database (if seed ran)
if db is not None:
try:
# Check if admin user exists
admin_user = db.query(User).filter(User.email == "admin@forgeflow.local").first()
if admin_user:
# Get the first API key for admin user
api_key_obj = db.query(ApiKey).filter(ApiKey.user_id == admin_user.id).first()
if api_key_obj:
return {
"email": admin_user.email,
"api_key": "⚠️ API key found in database but plaintext is not available. Check logs or seed again.",
"source": "database",
"message": "API key was seeded but plaintext is not stored. Check application logs for the original key."
}
except Exception as e:
pass
raise HTTPException(
status_code=404,
detail="Admin API key not found. Seed may not have run yet or encountered an error. Check application logs."
)
|