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.
2026-09-26 07:19:33 +00:00
parent 8a5ea57feb
commit 0c2a2275ce
3 changed files with 102 additions and 30 deletions
--- a/backend/.env.example
+++ b/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
--- a/backend/main.py
+++ b/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: