+ {/* Security Banner */}
+
+
+ SECURE SESSION - 2FA VERIFIED
+ |
+ User: {user?.username}
+
+
+ {/* Header */}
+
+
+
+ Secure Portal
+
+
+ Protected Dashboard
+
+
+
+
+
+ Standard Admin
+
+
+ End Session
+
+
+
+
+ {/* Stats Grid */}
+
+ {[
+ { label: 'Colony Members', val: stats.counters.users, icon: Users, color: 'text-coral' },
+ { label: 'Total Orders', val: stats.counters.orders, icon: TrendingUp, color: 'text-blue-400' },
+ { label: 'Revenue', val: `$${stats.counters.revenue.toFixed(2)}`, icon: DollarSign, color: 'text-yellow-400' },
+ { label: 'Products', val: stats.counters.products, icon: Package, color: 'text-purple-400' }
+ ].map((item, idx) => (
+
+
+
+
+
{item.label}
+
{item.val}
+
+ ))}
+
+
+ {/* Recent Orders */}
+ {stats.recent_orders && stats.recent_orders.length > 0 && (
+
+
Recent Orders
+
+
+
+
+ Order ID
+ Total
+ Status
+ Date
+
+
+
+ {stats.recent_orders.map((order) => (
+
+ #{order.id}
+ ${order.total.toFixed(2)}
+
+
+ {order.status}
+
+
+
+ {new Date(order.date).toLocaleDateString()}
+
+
+ ))}
+
+
+
+
+ )}
+
+ {/* Quick Actions */}
+
+
+
+
+
+
User Management
+
View and manage colony members
+
+
+
+
+
+
+
+
+
Product Inventory
+
Manage specimens
+
+
+
+
+
+ {/* Security Notice */}
+
+
+
+ Security Level: Maximum
+
+
+ This session is protected by two-factor authentication. All actions are logged and audited.
+ Your session will expire after 24 hours of inactivity.
+
+
+
+ );
+};
+
+export default SecurePortalDashboard;
diff --git a/src/frontend/src/pages/SecurePortalLogin.jsx b/src/frontend/src/pages/SecurePortalLogin.jsx
new file mode 100644
index 0000000..07faa30
--- /dev/null
+++ b/src/frontend/src/pages/SecurePortalLogin.jsx
@@ -0,0 +1,143 @@
+import React, { useState } from 'react';
+import { useNavigate, Link } from 'react-router-dom';
+import { User, Lock, ArrowRight, Bug, ShieldCheck } from 'lucide-react';
+
+const SecurePortalLogin = () => {
+ const navigate = useNavigate();
+ const [formData, setFormData] = useState({
+ username: '',
+ password: '',
+ totp_code: ''
+ });
+ const [error, setError] = useState('');
+ const [loading, setLoading] = useState(false);
+
+ const handleChange = (e) => {
+ const { name, value } = e.target;
+ if (name === 'totp_code') {
+ const sanitized = value.replace(/\D/g, '').slice(0, 6);
+ setFormData({ ...formData, [name]: sanitized });
+ } else {
+ setFormData({ ...formData, [name]: value });
+ }
+ };
+
+ const handleSubmit = async (e) => {
+ e.preventDefault();
+ setError('');
+ setLoading(true);
+
+ try {
+ const res = await fetch('/api/secure-portal/login', {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify(formData)
+ });
+
+ const data = await res.json();
+ if (res.ok) {
+ // Store secure portal credentials
+ localStorage.setItem('secure_token', data.access_token);
+ localStorage.setItem('secure_user', JSON.stringify(data.user));
+ // Also store as regular credentials so user is logged in site-wide
+ localStorage.setItem('token', data.access_token);
+ localStorage.setItem('user', JSON.stringify(data.user));
+ navigate('/secure-portal');
+ } else {
+ setError(data.detail || "Access denied. Invalid credentials or 2FA code.");
+ }
+ } catch (err) {
+ setError("The hive communication channel is blocked.");
+ } finally {
+ setLoading(false);
+ }
+ };
+
+ return (
+
+
+
+
+ {/* Progress indicator */}
+
+ {[1, 2, 3].map((s) => (
+
+
= s
+ ? 'bg-coral text-white'
+ : 'bg-hive-light/20 text-hive-subtle'
+ }`}>
+ {step > s ? : s}
+
+ {s < 3 &&
s ? 'bg-coral' : 'bg-hive-light/20'}`} />}
+
+ ))}
+
+
+ {/* Step 1: Introduction */}
+ {step === 1 && (
+
+
+
+
+
+
Set Up 2FA
+
+ Secure your admin access with two-factor authentication.
+ You'll need an authenticator app like Google Authenticator or Authy.
+
+
+
+ {error && (
+
+ {error}
+
+ )}
+
+
+ {loading ? "Generating..." : "Generate Secret Key"}
+
+
+
+
+ Back to Standard Admin
+
+
+ )}
+
+ {/* Step 2: Scan QR & Verify */}
+ {step === 2 && setupData && (
+
+
+
Scan QR Code
+
Open your authenticator app and scan this code
+
+
+ {/* QR Code */}
+
+
+
+
+ {/* Manual entry */}
+
+
+ Or enter manually:
+
+
+
+ {setupData.secret}
+
+
+ {copied ? : }
+
+
+
+
+ {/* Verification form */}
+
+
+ )}
+
+ {/* Step 3: Success */}
+ {step === 3 && (
+
+
+
+
+
+
2FA Enabled!
+
+ Your account is now protected with two-factor authentication.
+ You can now access the Secure Portal.
+
+
+
+
+
+ Important: Keep your authenticator app safe.
+ If you lose access, you may need to contact an administrator to reset your 2FA.
+
+
+
+
+ Go to Secure Portal Login
+
+
+ )}
+
+
+ );
+};
+
+export default SecurePortalSetup;
diff --git a/src/main.py b/src/main.py
index 8818b89..ed12f93 100644
--- a/src/main.py
+++ b/src/main.py
@@ -2,7 +2,7 @@
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
-from src.routes import catalog, cart, checkout, orders, auth, user, blog, review, forum, admin, health, redirect, debug
+from src.routes import catalog, cart, checkout, orders, auth, user, blog, review, forum, admin, health, redirect, debug, secure_portal
import os
from src.middleware.difficulty import DifficultyMiddleware
@@ -38,6 +38,7 @@
app.include_router(health.router, prefix="/api")
app.include_router(redirect.router, prefix="/api")
app.include_router(debug.router, prefix="/api")
+app.include_router(secure_portal.router, prefix="/api")
# Public config endpoint (non-sensitive settings for frontend)
@app.get("/api/config")
diff --git a/src/models.py b/src/models.py
index f7f4e04..58206c6 100644
--- a/src/models.py
+++ b/src/models.py
@@ -23,6 +23,10 @@ class User(Base):
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+ # TOTP/2FA fields for secure-portal
+ totp_secret = Column(String, nullable=True) # Base32 secret for TOTP
+ totp_enabled = Column(Boolean, default=False) # Whether 2FA is active
+
orders = relationship("Order", back_populates="user")
class Product(Base):
diff --git a/src/routes/secure_portal.py b/src/routes/secure_portal.py
new file mode 100644
index 0000000..ccae366
--- /dev/null
+++ b/src/routes/secure_portal.py
@@ -0,0 +1,374 @@
+"""
+Secure Portal - Admin with mandatory TOTP/2FA authentication.
+
+Coexists with /admin (no 2FA) for testing both authentication flows.
+
+Vulnerabilities planted:
+- V-030: No rate-limit on TOTP verification (brute force possible)
+- V-031: TOTP secret exposed in login response (info disclosure)
+"""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from pydantic import BaseModel
+from typing import Optional
+import pyotp
+import qrcode
+import io
+import base64
+
+from src.database import get_db
+from src.models import User, Order, Product
+from src.auth import (
+ create_access_token,
+ decode_access_token,
+ verify_password,
+ get_current_user,
+ SECRET_KEY,
+ ALGORITHM
+)
+from jose import jwt
+from datetime import datetime, timedelta
+
+router = APIRouter(prefix="/secure-portal", tags=["secure-portal"])
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
+
+# ============================================
+# Pydantic Models
+# ============================================
+
+class SecureLoginRequest(BaseModel):
+ username: str
+ password: str
+ totp_code: str # 6-digit code
+
+
+class EnableTOTPRequest(BaseModel):
+ totp_code: str # Verify user configured their app correctly
+
+
+# ============================================
+# Helper Functions
+# ============================================
+
+def generate_totp_secret() -> str:
+ """Generate a new TOTP base32 secret."""
+ return pyotp.random_base32()
+
+
+def verify_totp(secret: str, code: str) -> bool:
+ """
+ Verify a TOTP code.
+ V-030: NO rate limiting - allows brute force of 6-digit codes.
+ """
+ totp = pyotp.TOTP(secret)
+ return totp.verify(code)
+
+
+def create_secure_access_token(data: dict) -> str:
+ """
+ Create a JWT with '2fa_verified': True claim
+ to differentiate secure-portal tokens from regular ones.
+ """
+ to_encode = data.copy()
+ expire = datetime.utcnow() + timedelta(days=1)
+ to_encode.update({"exp": expire, "2fa_verified": True})
+ return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+# ============================================
+# Dependency: Verify token has 2FA claim
+# ============================================
+
+def get_current_user_2fa(
+ token: str = Depends(oauth2_scheme),
+ db: Session = Depends(get_db)
+) -> User:
+ """
+ Similar to get_current_user but REQUIRES the token
+ to have '2fa_verified': True claim.
+ """
+ payload = decode_access_token(token)
+ if not payload:
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Invalid credentials"
+ )
+
+ # Check 2FA claim
+ if not payload.get("2fa_verified"):
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="This endpoint requires 2FA authentication. Use /api/secure-portal/login"
+ )
+
+ username = payload.get("sub")
+ user = db.query(User).filter(User.username == username).first()
+ if not user:
+ raise HTTPException(status_code=404, detail="User not found")
+
+ return user
+
+
+def check_role_2fa(required_roles: list):
+ """
+ RBAC that REQUIRES 2FA authentication.
+ Only accepts tokens issued by /secure-portal/login.
+ """
+ def role_verifier(current_user: User = Depends(get_current_user_2fa)):
+ if current_user.role not in required_roles:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="Insufficient privileges for secure portal"
+ )
+ return current_user
+ return role_verifier
+
+
+# ============================================
+# Authentication Endpoints
+# ============================================
+
+@router.post("/login")
+def secure_login(data: SecureLoginRequest, db: Session = Depends(get_db)):
+ """
+ Login for secure-portal. Requires username, password AND TOTP code.
+
+ V-030: No rate-limit on TOTP verification (brute force possible).
+ V-031: TOTP secret exposed in response (info disclosure).
+ """
+ # Verify user and password
+ user = db.query(User).filter(User.username == data.username).first()
+
+ if not user or not verify_password(data.password, user.password_hash):
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Invalid credentials"
+ )
+
+ # Verify user has TOTP enabled
+ if not user.totp_enabled or not user.totp_secret:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="2FA not configured. Please set up TOTP first at /secure-portal/setup"
+ )
+
+ # V-030: NO rate limiting here - allows brute force of TOTP code
+ if not verify_totp(user.totp_secret, data.totp_code):
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Invalid TOTP code"
+ )
+
+ # Create token with 2FA claim
+ token = create_secure_access_token({
+ "sub": user.username,
+ "userId": user.id,
+ "role": user.role
+ })
+
+ # V-031: Expose totp_secret in response (info disclosure)
+ return {
+ "access_token": token,
+ "token_type": "bearer",
+ "user": {
+ "id": user.id,
+ "username": user.username,
+ "role": user.role,
+ "totp_secret": user.totp_secret # V-031: VULNERABILITY - Should not be here
+ },
+ "2fa_verified": True
+ }
+
+
+@router.post("/setup-totp")
+def setup_totp(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user)
+):
+ """
+ Generate a new TOTP secret and QR code for authenticator app setup.
+ (Google Authenticator, Authy, etc.)
+
+ Requires normal login but NOT prior 2FA.
+ """
+ # Only admin/staff can use secure-portal
+ if current_user.role not in ["admin", "staff"]:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="Only admin/staff can access secure portal"
+ )
+
+ # Generate new secret
+ secret = generate_totp_secret()
+
+ # Create provisioning URI for QR
+ totp = pyotp.TOTP(secret)
+ provisioning_uri = totp.provisioning_uri(
+ name=current_user.email,
+ issuer_name="BugStore Secure Portal"
+ )
+
+ # Generate QR code as base64
+ qr = qrcode.QRCode(version=1, box_size=10, border=5)
+ qr.add_data(provisioning_uri)
+ qr.make(fit=True)
+ img = qr.make_image(fill_color="black", back_color="white")
+
+ buffer = io.BytesIO()
+ img.save(buffer, format='PNG')
+ buffer.seek(0)
+ qr_base64 = base64.b64encode(buffer.getvalue()).decode()
+
+ # Save secret temporarily (not enabled yet)
+ current_user.totp_secret = secret
+ db.commit()
+
+ return {
+ "secret": secret,
+ "qr_code_base64": f"data:image/png;base64,{qr_base64}",
+ "provisioning_uri": provisioning_uri,
+ "message": "Scan the QR code with your authenticator app, then call /enable-totp with a valid code"
+ }
+
+
+@router.post("/enable-totp")
+def enable_totp(
+ data: EnableTOTPRequest,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user)
+):
+ """
+ Activate 2FA for the user after verifying they configured
+ their authenticator app correctly.
+ """
+ if not current_user.totp_secret:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="No TOTP secret found. Call /setup-totp first"
+ )
+
+ # Verify the code is valid
+ if not verify_totp(current_user.totp_secret, data.totp_code):
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="Invalid TOTP code. Make sure your authenticator is synced"
+ )
+
+ # Enable 2FA
+ current_user.totp_enabled = True
+ db.commit()
+
+ return {
+ "message": "2FA enabled successfully! You can now use /secure-portal/login",
+ "totp_enabled": True
+ }
+
+
+@router.post("/disable-totp")
+def disable_totp(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user_2fa)
+):
+ """
+ Disable 2FA for the user. Requires current 2FA authentication.
+ """
+ current_user.totp_enabled = False
+ current_user.totp_secret = None
+ db.commit()
+
+ return {
+ "message": "2FA disabled. You can no longer access the secure portal.",
+ "totp_enabled": False
+ }
+
+
+# ============================================
+# Protected Admin Endpoints (require 2FA)
+# ============================================
+
+@router.get("/stats")
+def get_secure_stats(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(check_role_2fa(["admin", "staff"]))
+):
+ """
+ Dashboard stats - REQUIRES 2FA authentication.
+ Same functionality as /admin/stats but with mandatory 2FA.
+ """
+ total_users = db.query(User).count()
+ total_orders = db.query(Order).count()
+ total_revenue = db.query(func.sum(Order.total)).scalar() or 0
+ total_products = db.query(Product).count()
+
+ recent_orders = db.query(Order).order_by(Order.created_at.desc()).limit(5).all()
+
+ return {
+ "counters": {
+ "users": total_users,
+ "orders": total_orders,
+ "revenue": total_revenue,
+ "products": total_products
+ },
+ "recent_orders": [
+ {
+ "id": o.id,
+ "total": o.total,
+ "status": o.status,
+ "date": o.created_at
+ } for o in recent_orders
+ ],
+ "security_level": "2FA_VERIFIED"
+ }
+
+
+@router.get("/users")
+def list_secure_users(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(check_role_2fa(["admin"]))
+):
+ """
+ List users - REQUIRES 2FA authentication and admin role.
+ """
+ users = db.query(User).all()
+ return [
+ {
+ "id": u.id,
+ "username": u.username,
+ "email": u.email,
+ "role": u.role,
+ "totp_enabled": u.totp_enabled,
+ "created_at": u.created_at
+ } for u in users
+ ]
+
+
+@router.get("/products")
+def list_secure_products(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(check_role_2fa(["admin", "staff"]))
+):
+ """
+ List products - REQUIRES 2FA authentication.
+ """
+ return db.query(Product).all()
+
+
+@router.get("/me")
+def get_secure_profile(
+ current_user: User = Depends(get_current_user_2fa)
+):
+ """
+ Get current user profile in secure session.
+ """
+ return {
+ "id": current_user.id,
+ "username": current_user.username,
+ "email": current_user.email,
+ "role": current_user.role,
+ "totp_enabled": current_user.totp_enabled,
+ "security_level": "2FA_VERIFIED"
+ }