Upload 2 files

src/api/admin.py

@@ -40,6 +40,15 @@ _CONFIG_DEFAULTS: Dict[str, Any] = {
 
 admin_sessions: Dict[str, float] = {}
 
+# ── Admin login lockout ────────────────────────────────────────────────────────
+# Tracks failed login attempts per IP.
+# After _MAX_LOGIN_ATTEMPTS failures within _LOCKOUT_WINDOW seconds,
+# that IP is locked out for _LOCKOUT_DURATION seconds.
+_login_attempts: Dict[str, list] = {}   # ip → [timestamp, ...]
+_MAX_LOGIN_ATTEMPTS  = 5
+_LOCKOUT_WINDOW      = 300   # 5 minutes — window in which attempts are counted
+_LOCKOUT_DURATION    = 900   # 15 minutes — lockout duration after max attempts
+
 
 def _sessions_file() -> str:
     cfg_path = os.environ.get("CONFIG_STORE_PATH", "")
@@ -138,16 +147,41 @@ class ConfigBulkUpdate(BaseModel):
 # ── Auth Endpoints ─────────────────────────────────────────────────────────────
 
 @router.post("/auth")
-async def admin_auth(req: LoginRequest):
+async def admin_auth(req: LoginRequest, request: Request):
     settings = get_settings()
+    client_ip = request.client.host if request.client else "unknown"
+    now = time.time()
+
+    # ── Lockout check ─────────────────────────────────────────────────────────
+    attempts = _login_attempts.get(client_ip, [])
+    # Purge attempts outside the lockout window
+    attempts = [t for t in attempts if now - t < _LOCKOUT_WINDOW]
+    if len(attempts) >= _MAX_LOGIN_ATTEMPTS:
+        wait = int(_LOCKOUT_DURATION - (now - attempts[0]))
+        raise HTTPException(
+            status_code=429,
+            detail=f"Too many failed login attempts. Try again in {wait} seconds."
+        )
+
+    # ── Password check ────────────────────────────────────────────────────────
     if not settings.admin_password or req.password != settings.admin_password:
-        raise HTTPException(status_code=401, detail="Invalid password")
+        attempts.append(now)
+        _login_attempts[client_ip] = attempts
+        remaining = _MAX_LOGIN_ATTEMPTS - len(attempts)
+        if remaining <= 0:
+            detail = f"Too many failed login attempts. Try again in {int(_LOCKOUT_DURATION / 60)} minutes."
+        else:
+            detail = f"Invalid password. {remaining} attempt{'s' if remaining != 1 else ''} remaining before lockout."
+        raise HTTPException(status_code=401, detail=detail)
+
+    # ── Success — clear attempts and issue session ─────────────────────────────
+    _login_attempts.pop(client_ip, None)
     token = secrets.token_hex(32)
     admin_sessions[token] = time.time() + 86400
     _save_sessions()
     resp = Response(content='{"status":"ok"}', media_type="application/json")
     resp.set_cookie(key="admin_session", value=token, httponly=True,
-                    max_age=86400, samesite="lax")
+                    max_age=86400, samesite="lax", secure=True)
     return resp
 
 
@@ -208,7 +242,7 @@ async function doLogin(e){
   const pw=document.getElementById('pw').value;
   const r=await fetch('/admin/auth',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({password:pw})});
   if(r.ok){window.location.href='/admin';}
-  else{document.getElementById('err').style.display='block';document.getElementById('pw').value='';document.getElementById('pw').focus();}
+  else{const data=await r.json().catch(()=>({}));const msg=data.detail||'Incorrect password. Try again.';const e=document.getElementById('err');e.textContent=msg;e.style.display='block';document.getElementById('pw').value='';document.getElementById('pw').focus();}
   return false;
 }
 </script></body></html>""")

src/api/agent.py

@@ -223,6 +223,39 @@ CURRENT_INFO_RE = re.compile(
     r"|news|update|updated|prime minister|ceo|minister|king|ruler|who won)\b",
     re.IGNORECASE,
 )
+
+# ── Academic guardrail — appended to all LLM system prompts ──────────────────
+_GUARDRAIL = (
+    "IMPORTANT — GUARDRAILS: "
+    "You must never make negative, critical, or controversial statements about "
+    "Khalifa University, its leadership, staff, or policies; the UAE government, "
+    "its leaders, laws, or institutions; or any religion. "
+    "If asked for opinions on these topics, politely decline and redirect to library services. "
+    "Never generate content that is politically sensitive, religiously offensive, "
+    "or inappropriate for a university academic environment. "
+    "You may answer general factual questions — such as historical facts, science, "
+    "geography, or current affairs — but always from a neutral, academic perspective. "
+    "Never produce content that could embarrass Khalifa University or its library."
+)
+
+# ── Prompt injection detection ────────────────────────────────────────────────
+_INJECTION_RE = re.compile(
+    r"\b(ignore\s+(previous|above|all|prior)\s+instructions?"
+    r"|disregard\s+(all|previous|above|prior)\s+instructions?"
+    r"|override\s+(system|instructions?|prompt)"
+    r"|new\s+instructions?\s*(are|follow|:|;)"
+    r"|you\s+are\s+now\s+(a|an|the|going)"
+    r"|act\s+as\s+(a|an|if|though|when)\b"
+    r"|act\s+like\s+(a|an)\s+\w"
+    r"|pretend\s+(you\s+are|to\s+be)"
+    r"|jailbreak|dan\s+mode|developer\s+mode|god\s+mode"
+    r"|system\s+prompt\s*(is|was|says|leak|show|reveal|print|output)"
+    r"|reveal\s+(your\s+)?(system\s+)?prompt"
+    r"|print\s+(your\s+)?(system\s+)?instructions?"
+    r"|forget\s+(you\s+are|your\s+role|everything)"
+    r"|do\s+anything\s+now|no\s+restrictions?)\b",
+    re.IGNORECASE,
+)
 SUMMARY_RE = re.compile(
     r"\b(summar(y|ize|ise|izing|ising)|overview|explain|what is|what are|describe"
     r"|tell me about|give me a summary|give me an overview|give me a brief"
@@ -1112,7 +1145,7 @@ async def _libbee_casual_response(question: str, history: List[ChatMessage], mod
         msgs = [{"role": "system", "content": (
             "You are LibBee, the Khalifa University Library AI Assistant. "
             "Respond warmly and naturally in 1-3 sentences as a friendly librarian. "
-            "No markdown, no bullet points."
+            "No markdown, no bullet points. " + _GUARDRAIL
         )}]
         _cfg = _get_runtime_config()
         _ci = _cfg.get("custom_instructions", "").strip()
@@ -1137,7 +1170,8 @@ async def _rag_answer(question: str, rag_results: List[dict], history: List[Chat
         "Be concise, friendly, and use HTML <br> for line breaks where needed. "
         f"If the context does not contain enough information, say so and suggest Ask a Librarian at {ASK_LIBRARIAN_URL}. "
         "Never invent facts or URLs.\n\n"
-        f"CONTEXT:\n{context}"
+        f"CONTEXT:\n{context}\n\n"
+        + _GUARDRAIL
     )
     _cfg = _get_runtime_config()
     _ci = _cfg.get("custom_instructions", "").strip()
@@ -1168,7 +1202,8 @@ async def _llm_general_answer(question: str, history: List[ChatMessage], model:
     system = (
         "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). "
         "Answer factually and concisely in 3-5 sentences. Use HTML <br> for line breaks. "
-        "KU means Khalifa University. If the topic is academic, briefly offer to help find library resources."
+        "KU means Khalifa University. If the topic is academic, briefly offer to help find library resources. "
+        + _GUARDRAIL
     )
     _cfg = _get_runtime_config()
     _ci = _cfg.get("custom_instructions", "").strip()
@@ -3765,6 +3800,25 @@ async def _agent_query_inner(
     start_time: float,
 ) -> AgentResponse:
 
+    # ── Prompt injection detection ────────────────────────────────────────────
+    # Fires before any processing. Catches attempts to override system prompt,
+    # jailbreak, or extract internal instructions.
+    if question and _INJECTION_RE.search(question):
+        _safe_metrics_bucket("intents", "injection_blocked")
+        return AgentResponse(
+            answer=(
+                "I'm sorry, but I can't process that request. "
+                "I'm LibBee, the KU Library AI Assistant — here to help with "
+                "library services, research, and academic resources. "
+                f'How can I help you today? <a href="{ASK_LIBRARIAN_URL}" target="_blank">Ask a Librarian</a>'
+            ),
+            intent="blocked",
+            tools_used=["injection_guard"],
+            model_used=model,
+            response_time=time.time() - start_time,
+        )
+    # ───────────────────────────────��─────────────────────────────────────────
+
     if not question:
         follow_up_question, follow_up_suggestions = _social_follow_up()
         return AgentResponse(