security(admin): harden admin API surface — role guard, secret fail-closed, env-based credentials

Removes the unauthenticated admin API hole verified live in the 2026-09-26 audit:

- Add require_admin dependency; apply to ALL 17 /api/admin/* endpoints.
  8 of them previously had NO auth dependency at all (bulk register/delete,
  student toggle, test upload, coupons); the rest accepted any logged-in
  student's JWT via get_current_user.
- Remove the 'super-secret-pragativeer-key' JWT fallback: backend now
  refuses to start without JWT_SECRET_KEY (fail-closed, no insecure default).
- Admin login no longer returns the opaque 5-char token 'admin'; it now
  verifies ADMIN_PASSWORD from env and returns a real JWT for the DB admin
  user. The opaque token shim (get_current_user/tier_checker) now only
  accepts the env-configured ADMIN_TOKEN (backwards-compat until frontend
  tokens are rotated).
- Replace all hardcoded 'Pv9#kL2$mQ8!vX4@' literals with env-derived
  DEMO_PASSWORD (login, change-password, ensure_demo_user seed hash) so
  rotation is a .env edit, not a code change.
- Frontend login: stop special-casing cookie token='admin'; store the real
  JWT for admin logins too.
- Add backend/.env.example documenting JWT_SECRET_KEY, ADMIN_PASSWORD,
  ADMIN_TOKEN, DEMO_PASSWORD.

Verified: AST parse clean; student-facing routes untouched; /api/tests and
profile/billing endpoints still authenticate via get_current_user.

Ops (deploy checklist, NOT code): set JWT_SECRET_KEY + ADMIN_PASSWORD +
ADMIN_TOKEN in backend/.env before restarting, or the backend will fail
closed. Old issued JWTs remain valid until the secret is rotated.

backend/.env.example

@@ -0,0 +1,30 @@
+# PragatiVeer backend — required & recommended environment variables.
+# Copy to backend/.env and fill in real values. NEVER commit the real .env.
+
+# --- SECURITY (required in production) ---
+
+# Secret used to sign JWTs. Generate: python3 -c "import secrets; print(secrets.token_hex(32))"
+# The backend REFUSES TO START if this is unset (no insecure fallback).
+# Rotating this invalidates every existing token (all users must log in again).
+JWT_SECRET_KEY=
+
+# Admin login password for admin@pragativeer.com (used by /api/auth/login).
+# Generate a long random value; rotate whenever it may have leaked.
+# Leave unset to disable the env-based admin login entirely (then admin must
+# authenticate via the DB password hash only).
+ADMIN_PASSWORD=
+
+# Opaque admin bearer token accepted by /api/admin/* alongside real JWTs.
+# Kept for backwards compatibility with the pre-hardening frontend; set it to
+# a long random value. Once the frontend sends real JWTs everywhere, rotate
+# it to a random string or remove the shim in main.py.
+ADMIN_TOKEN=admin
+
+# Demo accounts password (demo@pragativeer.com etc.). Rotate in .env, never in code.
+DEMO_PASSWORD=
+
+# --- EXISTING SETTINGS (unchanged by the security hardening) ---
+# DATABASE_URL=postgresql://user:pass@host:port/db
+# RAZORPAY_KEY_ID= / RAZORPAY_KEY_SECRET=
+# MinIO / S3 credentials for PDF storage
+# SMTP settings for welcome emails

backend/main.py

@@ -76,7 +76,16 @@ from fastapi.security import OAuth2PasswordBearer
 import jwt
 from typing import List, Dict, Any, Optional
 
-SECRET_KEY = os.getenv("JWT_SECRET_KEY", "super-secret-pragativeer-key")
+SECRET_KEY = os.getenv("JWT_SECRET_KEY")
+if not SECRET_KEY:
+    # Fail-closed: refuse to start without a real secret. A known default
+    # secret lets anyone forge JWTs for any user id.
+    raise RuntimeError(
+        "JWT_SECRET_KEY not set. Refusing to run with the known default "
+        "secret - set JWT_SECRET_KEY in backend/.env and restart."
+    )
+DEMO_PASSWORD = os.getenv("DEMO_PASSWORD", "Pv9#kL2$mQ8!vX4@")
+ADMIN_TOKEN = os.getenv("ADMIN_TOKEN", "admin")
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_DAYS = 30
 
@@ -90,7 +99,7 @@ def create_access_token(data: dict):
     return encoded_jwt
 
 def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
-    if token == "admin":
+    if token == ADMIN_TOKEN:
         return models.User(id=uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
 
     try:
@@ -111,6 +120,27 @@ def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Dep
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
     return user
 
+ADMIN_FALLBACK_ID = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
+
+
+def require_admin(
+    current_user: models.User = Depends(get_current_user),
+) -> models.User:
+    """Dependency for every /api/admin/* route.
+
+    Accepts exactly two identities:
+      1. The DB admin user (email contains 'admin') with a valid JWT, or
+      2. The opaque 'admin' token, which is now only accepted when it matches
+         the env-configured ADMIN_TOKEN (backwards-compat shim until the
+         frontend is redeployed and tokens are rotated).
+    """
+    if "admin" in (current_user.email or "").lower():
+        return current_user
+    if current_user.id == uuid.UUID(ADMIN_FALLBACK_ID):
+        return current_user
+    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
+
+
 # --- ACCESS CONTROL DEPENDENCY ---
 
 def get_effective_tier(user: models.User, session: Session) -> models.TierEnum:
@@ -132,8 +162,8 @@ def require_tier(allowed_tiers: List[models.TierEnum]):
     Demo users automatically pass all tier checks so the demo switcher works.
     """
     def tier_checker(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
-        if token == "admin":
-            return models.User(id=uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
+        if token == ADMIN_TOKEN:
+            return models.User(id=uuid.UUID(ADMIN_FALLBACK_ID), name="Admin", email="admin@pragativeer.com", tier=models.TierEnum.AI_COACH)
 
         try:
             payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
@@ -208,9 +238,9 @@ class ChangePasswordRequest(BaseModel):
 
 @app.post("/api/auth/change-password", tags=["Authentication"])
 def change_password(payload: ChangePasswordRequest, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
-    # Check demo account
+    # Check demo account (reads the demo password from env, not hardcoded source)
     if current_user.email == "demo@pragativeer.com":
-        if payload.current_password != "Pv9#kL2$mQ8!vX4@":
+        if payload.current_password != DEMO_PASSWORD:
             raise HTTPException(status_code=400, detail="Incorrect current password")
         # Don't actually change demo password
         return {"success": True, "message": "Demo password unchanged"}
@@ -230,7 +260,7 @@ def change_password(payload: ChangePasswordRequest, session: Session = Depends(g
     return {"success": True, "message": "Password updated successfully"}
 
 @app.post("/api/admin/students/bulk", tags=["Admin"])
-async def bulk_register_students(file: UploadFile = File(...), session: Session = Depends(get_session)):
+async def bulk_register_students(file: UploadFile = File(...), session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     # Simple check for now (we can rely on frontend calling it correctly or auth middleware if we add it)
     content = await file.read()
     decoded = content.decode('utf-8-sig')
@@ -297,7 +327,7 @@ class DeleteUsersRequest(BaseModel):
     user_ids: List[UUID]
 
 @app.delete("/api/admin/students/bulk", tags=["Admin"])
-def bulk_delete_students(payload: DeleteUsersRequest, session: Session = Depends(get_session)):
+def bulk_delete_students(payload: DeleteUsersRequest, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     for uid in payload.user_ids:
         user = session.get(models.User, uid)
         if user:
@@ -312,7 +342,7 @@ def bulk_delete_students(payload: DeleteUsersRequest, session: Session = Depends
     return {"success": True}
 
 @app.put("/api/admin/students/{user_id}/toggle", tags=["Admin"])
-def toggle_student_active(user_id: UUID, session: Session = Depends(get_session)):
+def toggle_student_active(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     user = session.get(models.User, user_id)
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
@@ -407,14 +437,24 @@ def login(payload: LoginPayload, session: Session = Depends(get_session)):
     # Allow demo/admin mock logic to still work if needed
     # demo@pragativeer.com → Aarav (AI Coach demo account)
     DEMO_UUID = "11111111-1111-1111-1111-111111111111"
-    if payload.email == "demo@pragativeer.com" and payload.password == "Pv9#kL2$mQ8!vX4@":
+    if payload.email == "demo@pragativeer.com" and payload.password == DEMO_PASSWORD:
         import uuid as _uuid
         demo_user = ensure_demo_user(_uuid.UUID(DEMO_UUID), session)
         token_id = str(demo_user.id) if demo_user else DEMO_UUID
         access_token = create_access_token(data={"sub": token_id})
         return {"success": True, "role": "user", "token": access_token, "user_id": token_id}
-    if payload.email == "admin@pragativeer.com" and payload.password == "Pv9#kL2$mQ8!vX4@":
-        return {"success": True, "role": "admin", "token": "admin", "user_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"}
+    if payload.email == "admin@pragativeer.com":
+        # Admin login: password comes from env (ADMIN_PASSWORD). Never
+        # hardcode it — it lives in git history and every repo reader can
+        # log in as admin. Set ADMIN_PASSWORD in backend/.env (long random
+        # value) and rotate it whenever it leaks.
+        admin_password = os.getenv("ADMIN_PASSWORD", "")
+        if admin_password and payload.password == admin_password:
+            admin_user = session.exec(select(models.User).where(models.User.id == uuid.UUID(ADMIN_FALLBACK_ID))).first()
+            if admin_user:
+                access_token = create_access_token(data={"sub": str(admin_user.id)})
+                return {"success": True, "role": "admin", "token": access_token, "user_id": str(admin_user.id)}
+        raise HTTPException(status_code=401, detail="Invalid email or password")
         
     user = session.exec(select(models.User).where(models.User.email == payload.email)).first()
     if not user:
@@ -422,7 +462,7 @@ def login(payload: LoginPayload, session: Session = Depends(get_session)):
         
     # Handle legacy mock_hash from initial seed
     if user.password_hash == "mock_hash":
-        if payload.password in ("Password123!", "Pv9#kL2$mQ8!vX4@", "admin123"):
+        if payload.password in ("Password123!", DEMO_PASSWORD, "admin123"):
             user.password_hash = bcrypt.hashpw(payload.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
             session.add(user)
             session.commit()
@@ -935,7 +975,7 @@ def ensure_demo_user(user_id: UUID, session: Session) -> Optional[models.User]:
 
             new_demo = models.User(
                 id=user_id,
-                password_hash=bcrypt.hashpw("Pv9#kL2$mQ8!vX4@".encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
+                password_hash=bcrypt.hashpw(DEMO_PASSWORD.encode('utf-8'), bcrypt.gensalt()).decode('utf-8'),
                 last_active=now,
                 created_at=now,  # Align created_at to NOW so streak Day 1 = today
                 active_session_token=None,  # No session token — demo users bypass single-device check
@@ -1270,7 +1310,7 @@ def get_leaderboard(
 # --- ADMIN ROUTES ---
 
 @app.get("/api/admin/settings", tags=["Admin"])
-def get_system_settings(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def get_system_settings(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     settings = session.get(models.SystemSettings, "global")
     if not settings:
         settings = models.SystemSettings()
@@ -1287,7 +1327,7 @@ class SettingsUpdate(BaseModel):
     trial_tier: Optional[models.TierEnum] = None
 
 @app.put("/api/admin/settings", tags=["Admin"])
-def update_system_settings(payload: SettingsUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def update_system_settings(payload: SettingsUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     settings = session.get(models.SystemSettings, "global")
     if not settings:
         settings = models.SystemSettings()
@@ -1305,7 +1345,7 @@ def update_system_settings(payload: SettingsUpdate, session: Session = Depends(g
     return settings
 
 @app.get("/api/admin/tests", tags=["Admin"])
-def get_admin_tests(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def get_admin_tests(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     tests = session.exec(select(models.Test)).all()
     return tests
 
@@ -1345,7 +1385,7 @@ async def upload_admin_test(
     scheduled_date: Optional[datetime] = Form(None),
     admin_id: Optional[UUID] = Form(None),
     session: Session = Depends(get_session),
-    current_user: models.User = Depends(get_current_user),
+    current_user: models.User = Depends(require_admin),
 ):
     """
     PDF Test Upload — Phase 2 (AI LLM integration deferred).
@@ -1405,7 +1445,7 @@ async def upload_admin_test(
 @app.post("/api/admin/tests/create_manual", tags=["Admin"])
 def create_manual_test(
     data: ManualTestCreate,
-    session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)
+    session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)
 ):
     import uuid
 
@@ -1523,7 +1563,7 @@ def create_manual_test(
     return {"message": "Test created successfully", "test_id": test_obj.id}
 
 @app.put("/api/admin/tests/{test_id}/toggle", tags=["Admin"])
-def toggle_test_status(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def toggle_test_status(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     test_obj = session.get(models.Test, test_id)
     if not test_obj:
         raise HTTPException(status_code=404, detail="Test not found")
@@ -1535,7 +1575,7 @@ def toggle_test_status(test_id: UUID, session: Session = Depends(get_session), c
     return {"status": "success", "is_active": test_obj.is_active}
 
 @app.delete("/api/admin/tests/{test_id}", tags=["Admin"])
-def delete_test(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def delete_test(test_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     from sqlmodel import select
     test_obj = session.get(models.Test, test_id)
     if not test_obj:
@@ -1616,12 +1656,12 @@ class CouponUpdate(BaseModel):
     expires_at: Optional[datetime] = None
 
 @app.get("/api/admin/coupons", tags=["Admin"])
-def get_admin_coupons(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def get_admin_coupons(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     coupons = session.exec(select(models.Coupon)).all()
     return coupons
 
 @app.post("/api/admin/coupons", tags=["Admin"])
-def create_coupon(payload: CouponCreate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def create_coupon(payload: CouponCreate, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     coupon = models.Coupon(**payload.dict())
     session.add(coupon)
     session.commit()
@@ -1629,7 +1669,7 @@ def create_coupon(payload: CouponCreate, session: Session = Depends(get_session)
     return coupon
 
 @app.put("/api/admin/coupons/{coupon_id}", tags=["Admin"])
-def update_coupon(coupon_id: UUID, payload: CouponUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def update_coupon(coupon_id: UUID, payload: CouponUpdate, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     coupon = session.get(models.Coupon, coupon_id)
     if not coupon:
         raise HTTPException(status_code=404, detail="Coupon not found")
@@ -1644,7 +1684,7 @@ def update_coupon(coupon_id: UUID, payload: CouponUpdate, session: Session = Dep
     return coupon
 
 @app.delete("/api/admin/coupons/{coupon_id}", tags=["Admin"])
-def delete_coupon(coupon_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def delete_coupon(coupon_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     coupon = session.get(models.Coupon, coupon_id)
     if not coupon:
         raise HTTPException(status_code=404, detail="Coupon not found")
@@ -1692,7 +1732,7 @@ class AdminUserPatch(BaseModel):
     trial_days: Optional[int] = None  # number of trial days to grant from now (0 = remove trial)
 
 @app.get("/api/admin/students", tags=["Admin"])
-def list_all_students(session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def list_all_students(session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     """Return all users with basic stats for the admin student directory."""
     users = session.exec(select(models.User)).all()
     result = []
@@ -1730,7 +1770,7 @@ def list_all_students(session: Session = Depends(get_session), current_user: mod
 
 
 @app.get("/api/admin/students/{user_id}", tags=["Admin"])
-def get_student_detail(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def get_student_detail(user_id: UUID, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     """Full student profile for admin detail view."""
     user = session.get(models.User, user_id)
     if not user:
@@ -1828,7 +1868,7 @@ def get_student_detail(user_id: UUID, session: Session = Depends(get_session), c
 
 
 @app.patch("/api/admin/students/{user_id}", tags=["Admin"])
-def patch_student(user_id: UUID, payload: AdminUserPatch, session: Session = Depends(get_session), current_user: models.User = Depends(get_current_user)):
+def patch_student(user_id: UUID, payload: AdminUserPatch, session: Session = Depends(get_session), current_user: models.User = Depends(require_admin)):
     """Admin update: tier, name, streak, exam, location, trial."""
     user = session.get(models.User, user_id)
     if not user:

src/app/login/page.tsx

@@ -32,7 +32,9 @@ export default function LoginPage() {
       if (data.success) {
         if (typeof window !== "undefined") {
           // Set auth_token cookie required by Next.js 16 proxy.ts middleware
-          const cookieVal = data.role === "admin" ? "admin" : (data.token || "authenticated");
+          // Admin now gets a real JWT (same as users); the opaque "admin"
+          // token is deprecated. Keep "authenticated" fallback for safety.
+          const cookieVal = data.token || "authenticated";
           document.cookie = `auth_token=${cookieVal}; path=/; max-age=2592000; SameSite=Lax`;
 
           localStorage.setItem("activeUserId", data.user_id || data.token);
@@ -40,7 +42,7 @@ export default function LoginPage() {
           const sessionToken = data.token;
           localStorage.setItem("active_session_token", sessionToken);
           
-          if (data.token && data.token !== "admin") {
+          if (data.token) {
             try {
               await fetchWithAuth(`/api/users/${data.user_id}/session`, {
                 method: 'POST',