""" IrisAuth Flask Backend ====================== All models are downloaded automatically from HuggingFace on first startup. Set the HF_TOKEN environment variable if your repo is private. """ import logging import os import traceback from datetime import datetime import base64 import pickle import io import numpy as np import torch import torchvision.utils as vutils from PIL import Image from flask import Flask, request, jsonify, render_template from flask_cors import CORS from scipy.stats import weibull_min from stats import update_login, update_registration, update_phase1, update_phase2, update_generation, get_dashboard_stats import time from collections import defaultdict from config import Config import json import uuid import os import sys # from gallery import save_gallery, load_gallery # from iris_recognition import embed_image, predict_hybrid, gallery, gallery_data, weibull_models # from iris_recognition import embed_image, predict_robust, register_person, gallery, weibull_models from models import ( load_biometric_model, load_phase2_model, load_gan_model, phase1_transform, pad_explainable, device, Generator, ) from utils import allowed, save_upload, to_base64, save_uploaded_file, allowed_file from iris_recognition import ( embed_image, predict_robust, register_person, gallery, weibull_models, mean_all, std_all, PKL_PATH, HF_FILENAME, HF_REPO_ID, sync_gallery_from_hf ) from huggingface_hub import HfApi logging.basicConfig( level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger(__name__) # ───────────────────────────────────────── # App init # ───────────────────────────────────────── app = Flask(__name__) CORS(app) # ───────────────────────────────────────── # Load models from HuggingFace # ───────────────────────────────────────── print("🚀 Loading models from HuggingFace…") # phase1_model = load_phase1_model() phase1_model, phase1_classes, phase1_thresholds, phase1_ckpt = load_biometric_model() print("✅ Phase 1 model ready") print("✅ Phase 1 model ready") phase2_model = load_phase2_model() print("✅ Phase 2 model ready") try: gan_model = load_gan_model(z_dim=Config.Z_DIM) print("✅ GAN model ready") except Exception as e: print(f"⚠️ GAN model not loaded: {e}") gan_model = None print(f"💻 Device: {device}") import requests as http_requests GROQ_API_KEY = os.getenv("GROQ_API_KEY") if not GROQ_API_KEY: print("⚠️ GROQ_API_KEY is not set!") CHAT_URL = "https://api.groq.com/openai/v1/chat/completions" CHAT_HEADERS = { "Authorization": f"Bearer {GROQ_API_KEY}", "Content-Type": "application/json", } MAX_ATTEMPTS = 3 LOCKOUT_SECONDS = 300 ADMIN_IDENTITIES = {"MMU_1"} LOGS_PATH = "models/login_logs.json" # PENDING_REQUESTS = {} # { request_id: { person_id, device_id, image_paths, phase1, phase2, status, created_at } } # PENDING_REQUESTS_FOLDER = "pending_registrations" PENDING_REQUESTS_FOLDER = "pending_registrations" PENDING_REQUESTS_PATH = "models/pending_requests.json" os.makedirs(PENDING_REQUESTS_FOLDER, exist_ok=True) # Always ensure logs file exists and is valid JSON on startup os.makedirs(os.path.dirname(LOGS_PATH), exist_ok=True) try: with open(LOGS_PATH, "r") as f: content = f.read().strip() if not content: raise ValueError("empty") json.loads(content) except: with open(LOGS_PATH, "w") as f: json.dump([], f) # Tracks global failures across ALL phases (Phase 1, Phase 2, Login) per session. # Once total failures reach MAX_ATTEMPTS, the session is locked out of every phase. lockout_store = defaultdict(lambda: {"attempts": 0, "locked_until": 0}) # def load_logs(): # if os.path.exists(LOGS_PATH): # with open(LOGS_PATH, "r") as f: # return json.load(f) # return [] def _do_register(person_id, saved_paths): """Shared registration logic used by /register and approve_request.""" if person_id in gallery: return None, f"Person '{person_id}' is already registered." if len(saved_paths) < 2: return None, "Need at least 2 valid iris images." result = register_person(person_id, saved_paths) update_registration(person_id) # Save to disk + HuggingFace with open(PKL_PATH, 'wb') as f: pickle.dump({ 'gallery': gallery, 'weibull_models': weibull_models, 'mean_all': mean_all, 'std_all': std_all }, f) try: api = HfApi() api.upload_file( path_or_fileobj=PKL_PATH, path_in_repo=HF_FILENAME, repo_id=HF_REPO_ID, repo_type="model" ) except Exception as e: print(f"⚠️ HF sync failed: {e}") return result, None # result, error def load_logs(): if os.path.exists(LOGS_PATH): try: with open(LOGS_PATH, "r") as f: content = f.read().strip() if not content: return [] return json.loads(content) except (json.JSONDecodeError, ValueError): # File is corrupted — reset it with open(LOGS_PATH, "w") as f: json.dump([], f) return [] return [] def load_pending_requests(): if os.path.exists(PENDING_REQUESTS_PATH): try: with open(PENDING_REQUESTS_PATH, "r") as f: content = f.read().strip() if not content: return {} return json.loads(content) except (json.JSONDecodeError, ValueError): return {} return {} def save_pending_requests(): with open(PENDING_REQUESTS_PATH, "w") as f: json.dump(PENDING_REQUESTS, f, indent=2) PENDING_REQUESTS = load_pending_requests() # ← loads from disk on startup def save_log_entry(entry): logs = load_logs() logs.append(entry) # Keep only last 500 entries so file doesn't grow forever logs = logs[-500:] with open(LOGS_PATH, "w") as f: json.dump(logs, f, indent=2) # def get_session_id(req): # return req.headers.get("X-Device-ID") or req.remote_addr or "unknown" def get_session_id(req): sid = req.headers.get("X-Device-ID") or req.remote_addr or "unknown" logger.info(f"🆔 SESSION_ID: '{sid}' | X-Device-ID: '{req.headers.get('X-Device-ID')}' | IP: '{req.remote_addr}'") return sid # def is_locked(session_id): # """Return (True, remaining_time_str) if the session is currently locked out.""" # # record = lockout_store[session_id] # record = lockout_store.get(session_id, {"attempts": 0, "locked_until": 0}) # if record["locked_until"] > time.time(): # remaining = int(record["locked_until"] - time.time()) # minutes = remaining // 60 # seconds = remaining % 60 # return True, f"{minutes}m {seconds}s" # return False, None def is_locked(session_id): """Return (True, remaining_time_str) if the session is currently locked out.""" # record = lockout_store.get(session_id, {"attempts": 0, "locked_until": 0}) record = lockout_store[session_id] if record["locked_until"] > time.time(): remaining = int(record["locked_until"] - time.time()) minutes = remaining // 60 seconds = remaining % 60 return True, f"{minutes}m {seconds}s" # ADD these 3 lines ↓ if session_id in lockout_store and lockout_store[session_id]["attempts"] >= MAX_ATTEMPTS: lockout_store[session_id]["attempts"] = 0 lockout_store[session_id]["locked_until"] = 0 return False, None # this line already exists, don't duplicate it # def record_failed_attempt(session_id, phase: str = ""): # """ # Increment the global failure counter for this session (shared across all phases). # If the total reaches MAX_ATTEMPTS the session is locked for LOCKOUT_SECONDS. # Returns True if the session just became locked. # """ # record = lockout_store[session_id] # record["attempts"] += 1 # print(f"⚠️ [{phase}] Failed attempt {record['attempts']}/{MAX_ATTEMPTS} for session {session_id}") # if record["attempts"] >= MAX_ATTEMPTS: # record["locked_until"] = time.time() + LOCKOUT_SECONDS # record["attempts"] = 0 # reset counter for the next cycle after lockout expires # print(f"🔒 Session {session_id} locked for {LOCKOUT_SECONDS}s after {MAX_ATTEMPTS} failures across all phases.") # return True # just got locked # return False def record_failed_attempt(session_id, phase: str = ""): """ Increment the global failure counter for this session (shared across all phases). If the total reaches MAX_ATTEMPTS the session is locked for LOCKOUT_SECONDS. Returns True if the session just became locked. """ record = lockout_store[session_id] record["attempts"] += 1 logger.info(f"⚠️ [{phase}] FAILED | session: '{session_id}' | attempts: {record['attempts']}/{MAX_ATTEMPTS}") print(f"⚠️ [{phase}] Failed attempt {record['attempts']}/{MAX_ATTEMPTS} for session {session_id}") if record["attempts"] >= MAX_ATTEMPTS: record["locked_until"] = time.time() + LOCKOUT_SECONDS # NOTE: Do NOT reset attempts here — keep it at MAX_ATTEMPTS while locked # so get_system_status() and the chatbot accurately report the locked state. # The counter is reset in is_locked() only after the lockout period expires. print(f"🔒 Session {session_id} locked for {LOCKOUT_SECONDS}s after {MAX_ATTEMPTS} failures across all phases.") return True # just got locked return False def record_success(session_id): """On a successful authentication, clear the failure counter.""" lockout_store[session_id] = {"attempts": 0, "locked_until": 0} def check_lockout_response(session_id): """ Helper used at the top of every phase route. Returns a Flask JSON response if locked, or None if the request may proceed. """ locked, remaining = is_locked(session_id) if locked: return jsonify( success=False, locked=True, message=( f"Access denied. Too many failed attempts across all phases. " f"Try again in {remaining}." ), remaining=remaining, ), 403 return None # def get_system_status(session_id): # """Builds a live system status string for the chatbot context.""" # locked, remaining = is_locked(session_id) # # attempts_used = lockout_store[session_id]["attempts"] # attempts_used = lockout_store.get(session_id, {"attempts": 0})["attempts"] # attempts_left = MAX_ATTEMPTS - attempts_used # status_lines = [] # # Lockout status # if locked: # status_lines.append(f"LOCKOUT: User is currently locked out. Time remaining: {remaining}.") # else: # status_lines.append(f"LOCKOUT: Not locked. Attempts used: {attempts_used}/{MAX_ATTEMPTS}. Attempts remaining: {attempts_left}.") # # Gallery status # status_lines.append(f"GALLERY: {len(gallery)} registered identities.") # # Model status # status_lines.append(f"PHASE1 MODEL: {'loaded' if phase1_model else 'NOT loaded'}.") # status_lines.append(f"PHASE2 MODEL: {'loaded' if phase2_model else 'NOT loaded'}.") # status_lines.append(f"GAN MODEL: {'loaded' if gan_model else 'NOT loaded'}.") # return "\n".join(status_lines) def get_system_status(session_id): """Builds a live system status string for the chatbot context.""" locked, remaining = is_locked(session_id) # ✅ Use .get() to avoid creating a phantom defaultdict entry # attempts_used = lockout_store.get(session_id, {"attempts": 0})["attempts"] attempts_used = lockout_store[session_id]["attempts"] attempts_left = MAX_ATTEMPTS - attempts_used status_lines = [] if locked: status_lines.append(f"LOCKOUT: User is currently locked out. Time remaining: {remaining}.") else: status_lines.append(f"LOCKOUT: Not locked. Attempts used: {attempts_used}/{MAX_ATTEMPTS}. Attempts remaining: {attempts_left}.") status_lines.append(f"GALLERY: {len(gallery)} registered identities.") status_lines.append(f"PHASE1 MODEL: {'loaded' if phase1_model else 'NOT loaded'}.") status_lines.append(f"PHASE2 MODEL: {'loaded' if phase2_model else 'NOT loaded'}.") status_lines.append(f"GAN MODEL: {'loaded' if gan_model else 'NOT loaded'}.") return "\n".join(status_lines) # ───────────────────────────────────────── # Knowledge base + helper functions # ───────────────────────────────────────── IRIS_KB = [ ("what is iris recognition", "Iris recognition is a biometric method that uses the unique patterns of the iris " "to identify individuals. It is highly accurate and stable throughout a person's life."), ("presentation attack pad spoof", "Presentation Attack Detection (PAD) identifies fake iris images presented to the sensor, " "such as printed photos, artificial eyes, or replay attacks on screens."), ("phase1 iris detection", "Phase 1 checks whether the uploaded image actually contains an iris. " "A classifier rejects non-iris images before further processing."), ("phase2 liveness real fake", "Phase 2 runs the PAD model with Grad-CAM to detect spoofing. " "It tells you whether the iris is from a live person or an attack."), ("gradcam heatmap explanation", "Grad-CAM generates a heatmap highlighting the image regions that influenced the model's " "decision, making the AI output explainable."), ("gan generate synthetic", "The GAN (Generative Adversarial Network) creates synthetic iris images for data " "augmentation or testing. It does not represent real individuals."), ("register enroll gallery", "Registration stores a person's iris embeddings in the gallery. " "At least 2 iris images are required. Weibull models are fitted per person for open-set recognition."), ("weibull open set unknown", "Weibull distribution models the tail of distance scores to detect unknown identities. " "If a probe is too far from all gallery entries, it is classified as unknown."), ("login verify identity score", "Login computes a cosine-distance embedding match against the gallery and returns " "the best identity with a confidence score. Low scores mean low confidence."), ("embedding feature vector", "An iris embedding is a compact numerical vector that encodes the texture of the iris. " "Similar irises produce vectors that are close in distance."), ] def search_kb(user_msg: str, top_k: int = 2) -> str: msg_lower = user_msg.lower() scored = [] for keywords, answer in IRIS_KB: score = sum(1 for kw in keywords.split() if kw in msg_lower) scored.append((score, answer)) scored.sort(key=lambda x: x[0], reverse=True) context_parts = [ans for sc, ans in scored[:top_k] if sc > 0] return " ".join(context_parts) if context_parts else "" def query_hf_model(prompt: str) -> str: try: resp = http_requests.post( CHAT_URL, headers=CHAT_HEADERS, json={ "model": "llama-3.1-8b-instant", "messages": [ { "role": "system", "content": ( "You are an expert assistant for an iris biometric authentication system. " "You have deep knowledge of biometrics, computer vision, deep learning, " "presentation attack detection, GANs, and iris recognition. " "Answer clearly and concisely. If the question is unrelated to biometrics " "or the system, politely say it is outside your area of expertise." ) }, { "role": "user", "content": prompt } ], "max_tokens": 300, "temperature": 0.7, }, timeout=30, ) print(f"Groq status: {resp.status_code}") print(f"Groq response: {resp.text[:300]}") data = resp.json() if "choices" in data and data["choices"]: return data["choices"][0]["message"]["content"].strip() if "error" in data: return f"Error: {data['error']['message']}" return "Could not generate a response. Please try again." except http_requests.exceptions.Timeout: return "Request timed out. Please try again." except Exception as e: print(f"⚠️ Groq chat error: {e}") return "Chatbot is temporarily unavailable. Please try again." def build_prompt(user_msg: str, context: str) -> str: if context: return f"Context:\n{context}\n\nQuestion: {user_msg}" return user_msg # def img_to_base64(path): # ← ADD THIS BLOCK HERE # if os.path.exists(path): # with open(path, 'rb') as f: # return base64.b64encode(f.read()).decode('utf-8') # return None def img_to_base64(path, max_size=(128, 128), quality=60): if os.path.exists(path): img = Image.open(path).convert("RGB") img.thumbnail(max_size) buf = io.BytesIO() img.save(buf, format="JPEG", quality=quality) return base64.b64encode(buf.getvalue()).decode("utf-8") return None # ───────────────────────────────────────── # Routes # ───────────────────────────────────────── @app.route("/") def index(): return render_template("index.html") @app.route('/register', methods=['POST']) def register(): person_id = request.form.get("person_id", "").strip() if not person_id: return jsonify(success=False, message="Person ID is required."), 400 files = request.files.getlist("images") saved_paths = [] try: for f in files: if not allowed(f.filename): continue path = save_upload(f) saved_paths.append(path) result, error = _do_register(person_id, saved_paths) if error: return jsonify(success=False, message=error), 409 return jsonify( success=True, message=f"'{person_id}' registered.", person_id=person_id, total_registered=len(gallery), debug_images={ # preprocessing steps (same as login) "initial_crop": img_to_base64("static/debug/1_initial_crop.png"), "final_input": img_to_base64("static/debug/3_final_input.png"), "reg_dark": img_to_base64("static/debug/reg_1_dark.png"), } ) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 finally: for p in saved_paths: if os.path.exists(p): os.remove(p) # ============================== # LOGIN (MULTI IMAGE) # ============================== @app.route('/login', methods=['POST']) def login_route(): """Phase 3 — Iris recognition (identity verification).""" session_id = get_session_id(request) is_registration_check = request.form.get("is_registration", "false").lower() == "true" # ── Global lockout check (shared across all phases) ── lockout_resp = check_lockout_response(session_id) if lockout_resp: return lockout_resp if "image" not in request.files: return jsonify(success=False, message="No image provided."), 400 f = request.files["image"] if not allowed(f.filename): return jsonify(success=False, message="Unsupported file type."), 400 try: path = save_upload(f) vectors = embed_image(path) # os.remove(path) if vectors is None: return jsonify(success=False, message="Could not process iris image."), 400 identity, cos, hyb = predict_robust(vectors) score = hyb if identity != "unknown" else cos update_login(success=(identity != "unknown"), score=float(score)) debug_images = { "initial_crop": img_to_base64("static/debug/1_initial_crop.png"), "normalized": img_to_base64("static/debug/2_normalized.png"), "final_input": img_to_base64("static/debug/3_final_input.png"), "tta_sharpened": img_to_base64("static/debug/tta_sharpened.png"), "tta_contrast": img_to_base64("static/debug/tta_high_contrast.png"), } if identity == "unknown": if not is_registration_check: # ✅ Only count failures during real logins just_locked = record_failed_attempt(session_id, phase="Login") else: just_locked = False attempts_used = lockout_store[session_id]["attempts"] remaining_attempts = MAX_ATTEMPTS - attempts_used if not just_locked else 0 save_log_entry({ "timestamp": datetime.now().isoformat(), "session_id": session_id, "outcome": "denied", "identity": "unknown", "role": None, "score": round(float(score), 4), "phase_failed": "Login" }) return jsonify( success=False, identity=None, message="Identity not recognized.", score=round(float(score), 4), locked=just_locked, debug_images=debug_images, attempts_remaining=remaining_attempts, lockout_message=( "System locked due to too many failed attempts across all phases. " "Try again in 5 mins." ) if just_locked else ( f"{remaining_attempts} attempt(s) remaining before lockout." ), ) # Successful login — clear the failure counter record_success(session_id) role = "admin" if identity in ADMIN_IDENTITIES else "user" save_log_entry({ "timestamp": datetime.now().isoformat(), "session_id": session_id, "outcome": "granted", "identity": identity, "role": role, "score": round(float(score), 4), "phase_failed": None }) return jsonify( success=True, identity=identity, role=role, message=f"Welcome, {identity}!", score=round(float(score), 4), attempts_remaining=MAX_ATTEMPTS, debug_images={ "initial_crop": img_to_base64("static/debug/1_initial_crop.png"), "normalized": img_to_base64("static/debug/2_normalized.png"), "final_input": img_to_base64("static/debug/3_final_input.png"), "tta_sharpened": img_to_base64("static/debug/tta_sharpened.png"), "tta_contrast": img_to_base64("static/debug/tta_high_contrast.png"), } ) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 # @app.route('/login', methods=['POST']) # def login_route(): # """Phase 3 — Iris recognition (identity verification).""" # session_id = get_session_id(request) # is_registration_check = request.form.get("is_registration", "false").lower() == "true" # # ── Global lockout check (shared across all phases) ── # lockout_resp = check_lockout_response(session_id) # if lockout_resp: # return lockout_resp # if "image" not in request.files: # return jsonify(success=False, message="No image provided."), 400 # f = request.files["image"] # if not allowed(f.filename): # return jsonify(success=False, message="Unsupported file type."), 400 # try: # path = save_upload(f) # vectors = embed_image(path) # # os.remove(path) # if vectors is None: # return jsonify(success=False, message="Could not process iris image."), 400 # identity, cos, hyb = predict_robust(vectors) # score = hyb if identity != "unknown" else cos # update_login(success=(identity != "unknown"), score=float(score)) # debug_images = { # "initial_crop": img_to_base64("static/debug/1_initial_crop.png"), # "normalized": img_to_base64("static/debug/2_normalized.png"), # "final_input": img_to_base64("static/debug/3_final_input.png"), # "tta_sharpened": img_to_base64("static/debug/tta_sharpened.png"), # "tta_contrast": img_to_base64("static/debug/tta_high_contrast.png"), # } # if identity == "unknown": # if not is_registration_check: # ✅ Only count failures during real logins # just_locked = record_failed_attempt(session_id, phase="Login") # else: # just_locked = False # just_locked = record_failed_attempt(session_id, phase="Login") # attempts_used = lockout_store[session_id]["attempts"] # remaining_attempts = MAX_ATTEMPTS - attempts_used if not just_locked else 0 # save_log_entry({ # "timestamp": datetime.now().isoformat(), # "session_id": session_id, # "outcome": "denied", # "identity": "unknown", # "role": None, # "score": round(float(score), 4), # "phase_failed": "Login" # }) # return jsonify( # success=False, # identity=None, # message="Identity not recognized.", # score=round(float(score), 4), # locked=just_locked, # debug_images=debug_images, # attempts_remaining=remaining_attempts, # lockout_message=( # "System locked due to too many failed attempts across all phases. " # "Try again in 5 mins." # ) if just_locked else ( # f"{remaining_attempts} attempt(s) remaining before lockout." # ), # ) # # Successful login — clear the failure counter # record_success(session_id) # role = "admin" if identity in ADMIN_IDENTITIES else "user" # save_log_entry({ # "timestamp": datetime.now().isoformat(), # "session_id": session_id, # "outcome": "granted", # "identity": identity, # "role": role, # "score": round(float(score), 4), # "phase_failed": None # }) # return jsonify( # success=True, # identity=identity, # role=role, # message=f"Welcome, {identity}!", # score=round(float(score), 4), # attempts_remaining=MAX_ATTEMPTS, # debug_images={ # "initial_crop": img_to_base64("static/debug/1_initial_crop.png"), # "normalized": img_to_base64("static/debug/2_normalized.png"), # "final_input": img_to_base64("static/debug/3_final_input.png"), # "tta_sharpened": img_to_base64("static/debug/tta_sharpened.png"), # "tta_contrast": img_to_base64("static/debug/tta_high_contrast.png"), # } # ) # except Exception as e: # traceback.print_exc() # return jsonify(success=False, message=str(e)), 500 def gallery_info(): return jsonify( total_persons=len(gallery), persons=list(gallery.keys()), ) @app.route("/stats") def stats(): return jsonify(get_dashboard_stats(gallery)) @app.route("/phase1", methods=["POST"]) def phase1(): """Phase 1 — Detect whether the image contains an iris.""" session_id = get_session_id(request) # ── Global lockout check ── lockout_resp = check_lockout_response(session_id) if lockout_resp: return lockout_resp if "image" not in request.files: return jsonify(success=False, message="No image provided"), 400 file = request.files["image"] if file.filename == "": return jsonify(success=False, message="Empty filename"), 400 filepath = save_uploaded_file(file) if not filepath: return jsonify(success=False, message="Invalid file type"), 400 try: image = Image.open(filepath).convert("RGB") input_tensor = phase1_transform(image).unsqueeze(0).to(device) # ── Same inference logic as second code ── with torch.no_grad(): logits_main, logits_eye, logits_disc, _ = phase1_model(input_tensor) probs_main = torch.softmax(logits_main, dim=1) eye_prob = torch.softmax(logits_eye, dim=1)[0, 0].item() disc_prob = torch.softmax(logits_disc, dim=1)[0, 0].item() probs = probs_main.clone() probs[0, 0] = 0.6 * probs_main[0, 0] + 0.4 * (eye_prob * disc_prob) probs = probs / probs.sum() confidence_tensor, pred_idx_tensor = probs.max(dim=1) pred_idx = pred_idx_tensor.item() confidence_val = confidence_tensor.item() confidence_score = float(confidence_val) * 100 pred_class_name = phase1_classes.get(pred_idx, "unknown") threshold = phase1_thresholds.get(pred_class_name, 0.5) accepted = confidence_val >= threshold # ── LOW CONFIDENCE ── if not accepted: just_locked = record_failed_attempt(session_id, phase="Phase1") attempts_used = lockout_store[session_id]["attempts"] remaining_attempts = MAX_ATTEMPTS - attempts_used if not just_locked else 0 return jsonify( success=True, is_iris=False, label="LOW_CONFIDENCE", confidence=round(confidence_score, 2), allow_phase2=False, locked=just_locked, attempts_remaining=remaining_attempts, message=( "System locked due to too many failures across all phases. Try again in 1 hour." if just_locked else "Low confidence score. Please upload a better image." ), ) # ── HUMAN IRIS ── if pred_class_name == "human_iris": update_phase1(is_iris=True) response = { "success": True, "is_iris": True, "label": "IRIS", "confidence": round(confidence_score, 2), "allow_phase2": True, "attempts_remaining": MAX_ATTEMPTS - lockout_store[session_id]["attempts"], "message": "Identity verified. Proceed to PAD check." } with Image.open(filepath) as img: response["image_base64"] = to_base64(img.convert("RGB")) return jsonify(response) # ── NON-IRIS CASES (match second logic) ── label_map = { "animal_eye": ("SUSPICIOUS_ANIMAL_EYE", "Suspicious input detected. Flagged for review."), "animals": ("NON_IRIS_ANIMAL", "Invalid input. Animal image detected."), "other": ("NON_IRIS_OTHER", "Invalid input. Please upload a clear human eye image."), } label, msg = label_map.get(pred_class_name, ("NON_IRIS", "Not an iris.")) update_phase1(is_iris=False) just_locked = record_failed_attempt(session_id, phase="Phase1") attempts_used = lockout_store[session_id]["attempts"] remaining_attempts = MAX_ATTEMPTS - attempts_used if not just_locked else 0 return jsonify( success=True, is_iris=False, label=label, confidence=round(confidence_score, 2), allow_phase2=False, locked=just_locked, attempts_remaining=remaining_attempts, message=( "System locked due to too many failures across all phases. Try again in 1 hour." if just_locked else msg ), ) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 finally: if os.path.exists(filepath): os.remove(filepath) @app.route("/phase2", methods=["POST"]) def phase2(): """Phase 2 — Presentation Attack Detection + Grad-CAM explanation.""" session_id = get_session_id(request) # ── Global lockout check (shared across all phases) ── lockout_resp = check_lockout_response(session_id) if lockout_resp: return lockout_resp if "image" not in request.files: return jsonify(success=False, message="No image provided"), 400 file = request.files["image"] if file.filename == "": return jsonify(success=False, message="Empty filename"), 400 filepath = save_uploaded_file(file) if not filepath: return jsonify(success=False, message="Invalid file type"), 400 try: result = pad_explainable(filepath, phase2_model) pred_label = result["prediction"] confidence = result["confidence"] * 100 heatmap_img = result["heatmap_image"] is_real = (pred_label == "REAL") update_phase2(is_real=is_real, confidence=result["confidence"]) # ── Record failure when a presentation attack is detected ── if not is_real: just_locked = record_failed_attempt(session_id, phase="Phase2") attempts_used = lockout_store[session_id]["attempts"] remaining_attempts = MAX_ATTEMPTS - attempts_used if not just_locked else 0 return jsonify({ "success": True, "is_real": False, "label": pred_label, "confidence": round(confidence, 2), "message": "Presentation attack detected", "human_explanation": result.get("human_explanation", ""), "region_explanation": result.get("region_explanation", ""), "heatmap_base64": to_base64(Image.fromarray(heatmap_img)), "locked": just_locked, "attempts_remaining": remaining_attempts, "lockout_message": ( "System locked due to too many failures across all phases. " "Try again in 1 hour." ) if just_locked else ( f"{remaining_attempts} attempt(s) remaining before lockout." ), }) return jsonify({ "success": True, "is_real": True, "label": pred_label, "confidence": round(confidence, 2), "message": "Authentication successful", "human_explanation": result.get("human_explanation", ""), "region_explanation": result.get("region_explanation", ""), "heatmap_base64": to_base64(Image.fromarray(heatmap_img)), "attempts_remaining": MAX_ATTEMPTS - lockout_store[session_id]["attempts"], }) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 finally: if os.path.exists(filepath): os.remove(filepath) @app.route("/generate", methods=["POST"]) def generate(): """Generate synthetic iris images using the GAN.""" if gan_model is None: return jsonify(success=False, message="GAN model not available"), 500 try: num_images = int(request.form.get("num_images", 1)) num_images = max(1, min(num_images, 10)) except (ValueError, TypeError): num_images = 1 generated_images = [] try: for i in range(num_images): noise = torch.randn(1, Config.Z_DIM, 1, 1).to(device) with torch.no_grad(): fake_img = gan_model(noise) img = (fake_img[0] + 1) / 2 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"gan_{timestamp}_{i}.png" save_path = os.path.join(Config.GENERATED_FOLDER, filename) vutils.save_image(img.cpu(), save_path) generated_images.append(f"/static/generated/{filename}") update_generation(len(generated_images)) return jsonify(success=True, num_images=num_images, images=generated_images) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 @app.route("/chat", methods=["POST"]) def chat(): """Iris biometrics chatbot powered by HuggingFace.""" data = request.get_json(silent=True) or {} user_msg = data.get("message", "").strip() if not user_msg: return jsonify(success=False, message="No message provided."), 400 session_id = get_session_id(request) logger.info(f"💬 CHAT | session: '{session_id}' | store: {dict(lockout_store[session_id])}") # Build live system context system_status = get_system_status(session_id) logger.info(f"💬 STATUS:\n{system_status}") kb_context = search_kb(user_msg) # Combine both into prompt full_context = f"LIVE SYSTEM STATUS:\n{system_status}" if kb_context: full_context += f"\n\nKNOWLEDGE BASE:\n{kb_context}" prompt = build_prompt(user_msg, full_context) reply = query_hf_model(prompt) return jsonify(success=True, reply=reply) # ───────────────────────────────────────── # ADMIN ROUTES # ───────────────────────────────────────── @app.route("/admin/gallery", methods=["GET"]) def admin_gallery(): """View all registered persons with embedding counts.""" gallery_info = {} for person_id, embeddings in gallery.items(): gallery_info[person_id] = { "embedding_count": len(embeddings), "role": "admin" if person_id in ADMIN_IDENTITIES else "user" } return jsonify( success=True, total_persons=len(gallery), gallery=gallery_info ) @app.route("/admin/delete/", methods=["DELETE"]) def admin_delete_person(person_id): """Delete a person from the gallery and sync to HuggingFace.""" global gallery, weibull_models if person_id not in gallery: return jsonify(success=False, message=f"'{person_id}' not found in gallery."), 404 # Prevent deleting the admin itself if person_id in ADMIN_IDENTITIES: return jsonify(success=False, message="Cannot delete an admin identity."), 403 # Remove from in-memory gallery del gallery[person_id] if person_id in weibull_models: del weibull_models[person_id] # Save locally with open(PKL_PATH, 'wb') as f: pickle.dump({ 'gallery': gallery, 'weibull_models': weibull_models, 'mean_all': mean_all, 'std_all': std_all }, f) # Push updated gallery to HuggingFace try: api = HfApi() api.upload_file( path_or_fileobj=PKL_PATH, path_in_repo=HF_FILENAME, repo_id=HF_REPO_ID, repo_type="model" ) hf_synced = True except Exception as e: print(f"❌ HF upload failed: {e}") hf_synced = False return jsonify( success=True, message=f"'{person_id}' deleted successfully.", hf_synced=hf_synced, total_persons=len(gallery) ) @app.route("/admin/logs", methods=["GET"]) def admin_logs(): logs = load_logs() return jsonify( success=True, total=len(logs), logs=logs[::-1] # newest first ) @app.route("/admin/logs/clear", methods=["DELETE"]) def admin_clear_logs(): with open(LOGS_PATH, "w") as f: json.dump([], f) return jsonify(success=True, message="Logs cleared.") @app.route("/admin/stats", methods=["GET"]) def admin_stats(): """Detailed system stats for admin dashboard.""" total_embeddings = sum(len(embs) for embs in gallery.values()) # Per-person embedding breakdown persons_detail = { pid: { "embeddings": len(embs), "role": "admin" if pid in ADMIN_IDENTITIES else "user" } for pid, embs in gallery.items() } # Active lockouts now = time.time() active_lockouts = [ { "session": sid, "remaining": int(info["locked_until"] - now) } for sid, info in lockout_store.items() if info["locked_until"] > now ] return jsonify( success=True, total_persons=len(gallery), total_embeddings=total_embeddings, active_lockouts=active_lockouts, lockout_count=len(active_lockouts), models={ "phase1": phase1_model is not None, "phase2": phase2_model is not None, "gan": gan_model is not None, }, persons=persons_detail ) @app.route("/admin/sync_gallery", methods=["POST"]) def admin_sync_gallery(): """Force re-sync gallery from HuggingFace (useful if gallery was updated externally).""" global gallery, weibull_models, mean_all, std_all try: synced_path = sync_gallery_from_hf() with open(synced_path, 'rb') as f: data = pickle.load(f) gallery = data['gallery'] weibull_models = data['weibull_models'] mean_all = data['mean_all'] std_all = data['std_all'] return jsonify( success=True, message="Gallery re-synced from HuggingFace.", total_persons=len(gallery) ) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 @app.route("/admin/unlock_session", methods=["POST"]) def admin_unlock_session(): """Manually unlock a locked-out session.""" data = request.get_json(silent=True) or {} session_id = data.get("session_id", "").strip() if not session_id: return jsonify(success=False, message="session_id is required."), 400 if session_id not in lockout_store: return jsonify(success=False, message="Session not found."), 404 lockout_store[session_id] = {"attempts": 0, "locked_until": 0} return jsonify( success=True, message=f"Session '{session_id}' unlocked successfully." ) @app.route("/admin/request_registration", methods=["POST"]) def request_registration(): try: person_id = request.form.get("person_id", "").strip() device_id = request.form.get("device_id", "").strip() if not person_id: return jsonify(success=False, message="person_id is required."), 400 files = request.files.getlist("images") if len(files) < 2: return jsonify(success=False, message="At least 2 images required."), 400 request_id = str(uuid.uuid4()) save_dir = os.path.join(PENDING_REQUESTS_FOLDER, request_id) os.makedirs(save_dir, exist_ok=True) saved_paths = [] for i, f in enumerate(files): filename = f"iris_{i}_{f.filename or 'image.jpg'}" save_path = os.path.join(save_dir, filename) f.save(save_path) saved_paths.append(save_path) PENDING_REQUESTS[request_id] = { "request_id": request_id, "person_id": person_id, "device_id": device_id, "image_paths": saved_paths, "status": "pending", "created_at": datetime.now().isoformat(), } save_pending_requests() # ← persist to disk print(f"✅ New registration request: {request_id} for {person_id}") return jsonify( success=True, request_id=request_id, message="Registration request submitted. Awaiting admin approval." ) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 @app.route("/admin/check_request/", methods=["GET"]) def check_request(request_id): """Flutter polls this to check if admin approved or rejected.""" req = PENDING_REQUESTS.get(request_id) if not req: return jsonify(success=False, message="Request not found."), 404 return jsonify( success=True, request_id=request_id, status=req["status"], # "pending" | "approved" | "rejected" person_id=req["person_id"], ) # ───────────────────────────────────────── # ROUTE 3: Admin panel lists all pending requests # GET /admin/pending_requests # ───────────────────────────────────────── @app.route("/admin/pending_requests", methods=["GET"]) def list_pending_requests(): """Admin panel fetches all pending registration requests.""" result = [] for rid, req in PENDING_REQUESTS.items(): result.append({ "request_id": rid, "person_id": req["person_id"], "device_id": req["device_id"], "status": req["status"], "created_at": req["created_at"], "image_count": len(req["image_paths"]), }) # newest first result.sort(key=lambda x: x["created_at"], reverse=True) return jsonify(success=True, total=len(result), requests=result) # ───────────────────────────────────────── # ROUTE 4: Admin approves → actually registers the person # POST /admin/approve_request/ # ───────────────────────────────────────── @app.route("/admin/approve_request/", methods=["POST"]) def approve_request(request_id): req = PENDING_REQUESTS.get(request_id) if not req: return jsonify(success=False, message="Request not found."), 404 if req["status"] != "pending": return jsonify(success=False, message=f"Already {req['status']}."), 400 try: result, error = _do_register(req["person_id"], req["image_paths"]) if error: PENDING_REQUESTS[request_id]["status"] = "rejected" return jsonify(success=False, message=error), 409 # PENDING_REQUESTS[request_id]["status"] = "approved" PENDING_REQUESTS[request_id]["status"] = "approved" save_pending_requests() # ← persist to disk # Clean up saved images for p in req["image_paths"]: if os.path.exists(p): os.remove(p) return jsonify( success=True, message=f"'{req['person_id']}' registered.", request_id=request_id, person_id=req["person_id"], total_registered=len(gallery), # ✅ exact same debug images as /register debug_images={ "initial_crop": img_to_base64("static/debug/1_initial_crop.png"), "final_input": img_to_base64("static/debug/3_final_input.png"), "reg_dark": img_to_base64("static/debug/reg_1_dark.png"), } ) except Exception as e: traceback.print_exc() return jsonify(success=False, message=str(e)), 500 @app.route("/admin/reject_request/", methods=["POST"]) def reject_request(request_id): """Admin rejects the pending registration request.""" req = PENDING_REQUESTS.get(request_id) if not req: return jsonify(success=False, message="Request not found."), 404 if req["status"] != "pending": return jsonify(success=False, message=f"Request is already {req['status']}."), 400 # PENDING_REQUESTS[request_id]["status"] = "rejected" PENDING_REQUESTS[request_id]["status"] = "rejected" save_pending_requests() # ← persist to disk print(f"❌ Rejected registration: {req['person_id']} (request {request_id})") return jsonify( success=True, message="Request rejected.", request_id=request_id, ) # if __name__=="__main__": # app.run(debug=True,host="0.0.0.0",port=5000) if __name__ == "__main__": app.run(host="0.0.0.0", port=7860)