Format source

Author: pablogsal

backend/app/admin_auth.py

@@ -23,8 +23,10 @@ async def create_admin_session(
 ) -> str:
     """Create a new admin session for a GitHub user."""
     session_token = secrets.token_urlsafe(48)
-    expires_at = datetime.now(UTC).replace(tzinfo=None) + timedelta(hours=duration_hours)
-    
+    expires_at = datetime.now(UTC).replace(tzinfo=None) + timedelta(
+        hours=duration_hours
+    )
+
     admin_session = AdminSession(
         session_token=session_token,
         github_user_id=github_user.id,
@@ -34,11 +36,11 @@ async def create_admin_session(
         github_avatar_url=github_user.avatar_url,
         expires_at=expires_at,
     )
-    
+
     db.add(admin_session)
     await db.commit()
     await db.refresh(admin_session)
-    
+
     return session_token
 
 
@@ -80,10 +82,10 @@ async def cleanup_expired_sessions(db: AsyncSession) -> None:
         )
     )
     expired_sessions = result.scalars().all()
-    
+
     for session in expired_sessions:
         session.is_active = False
-    
+
     if expired_sessions:
         await db.commit()
 
@@ -103,7 +105,7 @@ async def require_admin_auth(
             detail="Admin authentication required",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    
+
     try:
         session = await get_admin_session(db, admin_session_token)
     except Exception as e:
@@ -114,14 +116,14 @@ async def require_admin_auth(
             detail="Authentication service unavailable",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    
+
     if not session:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid or expired admin session",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    
+
     # Check if user is still an admin (for existing sessions, we need a valid token)
     # Note: This check is disabled for existing sessions as we don't store the access token
     # Team membership is verified during initial login only
@@ -131,7 +133,7 @@ async def require_admin_auth(
     #         status_code=status.HTTP_403_FORBIDDEN,
     #         detail="Admin privileges revoked",
     #     )
-    
+
     return session
 
 
@@ -146,7 +148,7 @@ async def optional_admin_auth(
     """
     if not admin_session_token:
         return None
-    
+
     try:
         return await require_admin_auth(request, admin_session_token, db)
     except HTTPException:

backend/app/auth.py

@@ -48,7 +48,9 @@ async def get_current_token(
     if not auth_token:
         # Only log token length and first/last 4 characters for security
         masked_token = f"{token[:4]}...{token[-4:]}" if len(token) > 8 else "***"
-        logger.warning(f"Authentication failed: Invalid token {masked_token} (length: {len(token)})")
+        logger.warning(
+            f"Authentication failed: Invalid token {masked_token} (length: {len(token)})"
+        )
         raise authentication_failed("Invalid or expired token")
 
     # Set user context for logging
@@ -58,4 +60,4 @@ async def get_current_token(
     await crud.update_token_last_used(db, token)
 
     logger.info(f"Authentication successful for token: {auth_token.name}")
-    return auth_token
+    return auth_token

backend/app/config.py

@@ -26,11 +26,17 @@ class Settings(BaseSettings):
     database_pool_recycle: int = 3600
     database_echo: bool = False
 
-    # CORS Configuration  
+    # CORS Configuration
     cors_origins: str = ""  # Comma-separated string, will be parsed to list
     cors_allow_credentials: bool = True
     cors_allow_methods: List[str] = ["GET", "POST", "PUT", "DELETE", "OPTIONS"]
-    cors_allow_headers: List[str] = ["Authorization", "Content-Type", "Accept", "Origin", "X-Requested-With"]
+    cors_allow_headers: List[str] = [
+        "Authorization",
+        "Content-Type",
+        "Accept",
+        "Origin",
+        "X-Requested-With",
+    ]
 
     # Pagination Configuration
     default_page_size: int = 100
@@ -41,18 +47,20 @@ class Settings(BaseSettings):
     # Authentication Configuration
     token_length: int = 32  # in bytes, results in 64 hex characters
     token_cleanup_days: int = 90
-    
+
     # GitHub OAuth Configuration
     github_client_id: str = ""
     github_client_secret: str = ""
     oauth_redirect_uri: str = "http://localhost:3000/admin/auth/callback"
     oauth_state_secret: str = "your-secret-key-change-me"
-    
+
     # Admin authorization via GitHub usernames
     admin_initial_username: str = ""  # Initial admin username (e.g., "pablogsal")
     # Legacy team-based auth (deprecated)
-    admin_github_org: str = ""  # GitHub organization name (e.g., "python") 
-    admin_github_teams: str = ""  # Comma-separated list of team slugs (e.g., "memory-python-org")
+    admin_github_org: str = ""  # GitHub organization name (e.g., "python")
+    admin_github_teams: str = (
+        ""  # Comma-separated list of team slugs (e.g., "memory-python-org")
+    )
 
     # Performance Configuration
     top_functions_limit: int = 10
@@ -78,14 +86,18 @@ def cors_origins_list(self) -> List[str]:
         """Parse CORS origins string into a list."""
         if not self.cors_origins.strip():
             return []
-        return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()]
-    
+        return [
+            origin.strip() for origin in self.cors_origins.split(",") if origin.strip()
+        ]
+
     @property
     def admin_github_teams_list(self) -> List[str]:
         """Parse admin GitHub teams string into a list."""
         if not self.admin_github_teams.strip():
             return []
-        return [team.strip() for team in self.admin_github_teams.split(",") if team.strip()]
+        return [
+            team.strip() for team in self.admin_github_teams.split(",") if team.strip()
+        ]
 
 
 @lru_cache()

backend/app/crud.py

@@ -99,7 +99,7 @@ async def create_commit(
     timestamp = commit.timestamp
     if timestamp.tzinfo is not None:
         timestamp = timestamp.replace(tzinfo=None)
-    
+
     db_commit = models.Commit(
         sha=commit.sha,
         timestamp=timestamp,
@@ -118,8 +118,7 @@ async def create_commit(
 async def get_binaries(db: AsyncSession) -> List[models.Binary]:
     result = await db.execute(
         select(models.Binary).order_by(
-            models.Binary.display_order.asc(), 
-            models.Binary.name.asc()
+            models.Binary.display_order.asc(), models.Binary.name.asc()
         )
     )
     return result.scalars().all()
@@ -217,7 +216,7 @@ async def get_runs_with_commits(
 
     if commit_sha:
         # Use prefix matching (starts with) for commit SHA
-        query = query.where(models.Run.commit_sha.ilike(f'{commit_sha}%'))
+        query = query.where(models.Run.commit_sha.ilike(f"{commit_sha}%"))
     if binary_id:
         query = query.where(models.Run.binary_id == binary_id)
     if environment_id:
@@ -239,7 +238,7 @@ async def count_runs(
 
     if commit_sha:
         # Use prefix matching (starts with) for commit SHA
-        query = query.where(models.Run.commit_sha.ilike(f'{commit_sha}%'))
+        query = query.where(models.Run.commit_sha.ilike(f"{commit_sha}%"))
     if binary_id:
         query = query.where(models.Run.binary_id == binary_id)
     if environment_id:
@@ -254,7 +253,7 @@ async def create_run(db: AsyncSession, run: schemas.RunCreate) -> models.Run:
     timestamp = run.timestamp
     if timestamp.tzinfo is not None:
         timestamp = timestamp.replace(tzinfo=None)
-    
+
     db_run = models.Run(
         run_id=run.run_id,
         commit_sha=run.commit_sha,
@@ -467,9 +466,7 @@ async def create_auth_token(
     db: AsyncSession, token: str, name: str, description: str = None
 ) -> models.AuthToken:
     """Create a new auth token."""
-    db_token = models.AuthToken(
-        token=token, name=name, description=description
-    )
+    db_token = models.AuthToken(token=token, name=name, description=description)
     db.add(db_token)
     await db.commit()
     await db.refresh(db_token)
@@ -512,30 +509,34 @@ async def deactivate_auth_token(db: AsyncSession, token_id: int) -> bool:
 async def get_admin_users(db: AsyncSession) -> List[models.AdminUser]:
     """Get all admin users."""
     result = await db.execute(
-        select(models.AdminUser).where(models.AdminUser.is_active == True).order_by(models.AdminUser.added_at)
+        select(models.AdminUser)
+        .where(models.AdminUser.is_active == True)
+        .order_by(models.AdminUser.added_at)
     )
     return result.scalars().all()
 
 
-async def get_admin_user_by_username(db: AsyncSession, username: str) -> Optional[models.AdminUser]:
+async def get_admin_user_by_username(
+    db: AsyncSession, username: str
+) -> Optional[models.AdminUser]:
     """Get admin user by GitHub username."""
     result = await db.execute(
         select(models.AdminUser).where(
             and_(
                 models.AdminUser.github_username == username,
-                models.AdminUser.is_active == True
+                models.AdminUser.is_active == True,
             )
         )
     )
     return result.scalars().first()
 
 
-async def create_admin_user(db: AsyncSession, username: str, added_by: str, notes: Optional[str] = None) -> models.AdminUser:
+async def create_admin_user(
+    db: AsyncSession, username: str, added_by: str, notes: Optional[str] = None
+) -> models.AdminUser:
     """Create a new admin user."""
     admin_user = models.AdminUser(
-        github_username=username,
-        added_by=added_by,
-        notes=notes
+        github_username=username, added_by=added_by, notes=notes
     )
     db.add(admin_user)
     await db.commit()
@@ -566,10 +567,12 @@ async def ensure_initial_admin(db: AsyncSession, username: str) -> None:
     """Ensure the initial admin user exists."""
     if not username:
         return
-    
+
     existing = await get_admin_user_by_username(db, username)
     if not existing:
         logger.info(f"Creating initial admin user: {username}")
-        await create_admin_user(db, username, "system", "Initial admin from environment variable")
+        await create_admin_user(
+            db, username, "system", "Initial admin from environment variable"
+        )
     else:
         logger.info(f"Initial admin user already exists: {username}")

backend/app/database.py

@@ -7,10 +7,11 @@
 
 logger = logging.getLogger(__name__)
 
+
 def create_database_engine():
     """Create database engine with proper configuration."""
     settings = get_settings()
-    
+
     engine_kwargs = {
         "echo": settings.database_echo,
         "pool_size": settings.database_pool_size,
@@ -20,7 +21,7 @@ def create_database_engine():
         # Add connection timeout for production
         "connect_args": {"timeout": 30} if "sqlite" in settings.database_url else {},
     }
-    
+
     # Add additional PostgreSQL/MySQL specific optimizations
     if "postgresql" in settings.database_url:
         engine_kwargs["connect_args"] = {
@@ -34,7 +35,7 @@ def create_database_engine():
             "charset": "utf8mb4",
             "autocommit": False,
         }
-    
+
     logger.info(
         f"Creating database engine: {settings.database_url.split('@')[-1] if '@' in settings.database_url else settings.database_url}"
     )
@@ -43,9 +44,10 @@ def create_database_engine():
         f"max_overflow={settings.database_max_overflow}, "
         f"recycle={settings.database_pool_recycle}s"
     )
-    
+
     return create_async_engine(settings.database_url, **engine_kwargs)
 
+
 # Create engine using factory function
 engine = create_database_engine()
 AsyncSessionLocal = async_sessionmaker(
@@ -87,6 +89,7 @@ async def drop_tables():
 from contextlib import asynccontextmanager
 from typing import AsyncGenerator
 
+
 @asynccontextmanager
 async def transaction_scope() -> AsyncGenerator[AsyncSession, None]:
     """