#!/usr/bin/env python3 import base64, hashlib, hmac, json, os, re, secrets, socket, sqlite3, subprocess, threading, time, urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer INTERFACE = os.environ.get('AP_INTERFACE', 'wls16b1') HOST, PORT = '0.0.0.0', 8080 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DB_FILE = os.environ.get('AP_DB_FILE', os.path.join(BASE_DIR, 'ap-dashboard.sqlite3')) CONNECTION_FILE = os.environ.get('AP_CONNECTION_FILE', os.path.join(BASE_DIR, 'ap-dashboard-connection')) UPTIME_FILE = os.environ.get('AP_UPTIME_FILE', os.path.join(BASE_DIR, 'ap-dashboard-uptime')) # Auto-migration de l'ancienne DB si présente OLD_DB_FILE = '/var/lib/ap-dashboard.sqlite3' if not os.path.exists(DB_FILE) and os.path.exists(OLD_DB_FILE): try: import shutil shutil.copy2(OLD_DB_FILE, DB_FILE) except Exception: pass SESSION_TTL = 3600 sessions = {} last_macs = set() def uptime_start(): try: return float(open(UPTIME_FILE, encoding='utf8').read().strip()) except (OSError, ValueError): return None def mark_ap_started(): try: folder = os.path.dirname(UPTIME_FILE) if folder: os.makedirs(folder, exist_ok=True) open(UPTIME_FILE, 'w', encoding='utf8').write(str(time.time())) except OSError: pass def ap_uptime(): started = uptime_start() return max(0, int(time.time() - started)) if started else 0 def device_name(ip, mac): mac = (mac or '').upper() # Les baux dnsmasq peuvent être dans plusieurs fichiers selon NetworkManager. lease_files = ['/var/lib/NetworkManager/dnsmasq.leases', '/var/lib/misc/dnsmasq.leases'] try: import glob lease_files += glob.glob('/var/lib/NetworkManager/dnsmasq-*.leases') for lease_file in lease_files: with open(lease_file, encoding='utf8') as f: for line in f: fields = line.strip().split() if len(fields) >= 4 and (fields[1].upper() == mac or fields[2] == ip): hostname = fields[3].strip() if hostname and hostname != '*': return hostname except (OSError, UnicodeError): pass # NetworkManager/dnsmasq écrit parfois uniquement le DHCPACK dans le journal. try: journal = command('journalctl','-u','NetworkManager','--no-pager','-n','500') pattern = r'DHCPACK\([^)]*\)\s+(\S+)\s+([0-9a-f:]{17})\s+(\S+)' for line in reversed(journal.splitlines()): match = re.search(pattern, line, re.I) if match and (match.group(1) == ip or match.group(2).upper() == mac): hostname = match.group(3).strip() if hostname and hostname != '*': return hostname except Exception: pass if ip and ip not in ('Non attribuée', '—'): try: name = socket.gethostbyaddr(ip)[0] if name and name != ip: return name.split('.')[0] except (socket.herror, socket.gaierror, OSError): pass host_line = command('ip','neigh','show',ip) match = re.search(r'\(([^)]+)\)', host_line) if match and match.group(1) != ip: return match.group(1) try: c = db(); row = c.execute('SELECT name FROM client_names WHERE mac=?', (mac,)).fetchone(); c.close() if row and row[0]: return row[0] except sqlite3.Error: pass return 'Client ' + mac.replace(':', '')[-4:] if mac else 'Nom non communiqué' # __AUTHENTICATION_IMPLEMENTATION__ def db(): folder = os.path.dirname(DB_FILE) if folder: os.makedirs(folder, exist_ok=True) c = sqlite3.connect(DB_FILE) c.execute('CREATE TABLE IF NOT EXISTS auth_users (username TEXT PRIMARY KEY, password_hash TEXT NOT NULL)') columns = {row[1] for row in c.execute('PRAGMA table_info(auth_users)').fetchall()} if 'totp_secret' not in columns: c.execute('ALTER TABLE auth_users ADD COLUMN totp_secret TEXT DEFAULT NULL') c.commit() recovery_table(c) c.execute('CREATE TABLE IF NOT EXISTS connection_logs(id INTEGER PRIMARY KEY,event_time TEXT,event TEXT,name TEXT,ip TEXT,mac TEXT)') c.execute('CREATE TABLE IF NOT EXISTS blocked_clients(mac TEXT PRIMARY KEY,blocked_at TEXT)') c.execute('CREATE TABLE IF NOT EXISTS client_names(mac TEXT PRIMARY KEY,name TEXT NOT NULL,last_ip TEXT,last_seen TEXT)') c.execute('CREATE TABLE IF NOT EXISTS preferences(key TEXT PRIMARY KEY, value TEXT NOT NULL)') c.commit(); return c def password_hash(value, salt=None): salt = salt or secrets.token_bytes(16) digest = hashlib.pbkdf2_hmac('sha256', value.encode(), salt, 240000) return salt.hex() + ':' + digest.hex() def password_ok(value, stored): try: salt, digest = stored.split(':', 1) actual = hashlib.pbkdf2_hmac('sha256', value.encode(), bytes.fromhex(salt), 240000).hex() return hmac.compare_digest(actual, digest) except (ValueError, TypeError): return False def totp_code(secret, counter=None): counter = int(time.time() // 30) if counter is None else counter key = base64.b32decode(secret + '=' * (-len(secret) % 8), casefold=True) raw = hmac.new(key, counter.to_bytes(8, 'big'), hashlib.sha1).digest() offset = raw[-1] & 15 number = int.from_bytes(raw[offset:offset + 4], 'big') & 0x7fffffff return f'{number % 1000000:06d}' def totp_ok(secret, code): code = str(code or '').strip() if not re.fullmatch(r'\d{6}', code): return False counter = int(time.time() // 30) return any(hmac.compare_digest(totp_code(secret, counter + drift), code) for drift in (-1, 0, 1)) def auth_state(): c = db(); row = c.execute('SELECT username,totp_secret FROM auth_users LIMIT 1').fetchone() rows = dict(c.execute("SELECT key,value FROM preferences").fetchall()); c.close() return { 'configured': bool(row), 'two_factor': bool(row and row[1]), 'app_name': rows.get('app_name', 'Administration de l’AP Wi‑Fi'), 'app_short_name': rows.get('app_short_name', 'AP Manageur') } RECOVERY_COUNT = 8 def recovery_table(c): c.execute('CREATE TABLE IF NOT EXISTS recovery_codes(code_hash TEXT PRIMARY KEY, used INTEGER NOT NULL DEFAULT 0)') def recovery_hash(code): return hashlib.sha256(re.sub(r'[^A-Z0-9]', '', str(code or '').upper()).encode()).hexdigest() def recovery_generate(): alphabet = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789' codes = [] for _ in range(RECOVERY_COUNT): raw = ''.join(secrets.choice(alphabet) for _ in range(10)) codes.append(raw[:5] + '-' + raw[5:]) c = db(); recovery_table(c); c.execute('DELETE FROM recovery_codes') c.executemany('INSERT INTO recovery_codes(code_hash) VALUES(?)', [(recovery_hash(x),) for x in codes]); c.commit(); c.close() return codes def recovery_use(code): cleaned = re.sub(r'[^A-Za-z0-9]', '', str(code or '')) if len(cleaned) != 10: return False c = db(); recovery_table(c) cur = c.execute('UPDATE recovery_codes SET used=1 WHERE code_hash=? AND used=0', (recovery_hash(cleaned),)) c.commit(); ok = cur.rowcount == 1; c.close(); return ok def recovery_remaining(): c = db(); recovery_table(c); n = c.execute('SELECT COUNT(*) FROM recovery_codes WHERE used=0').fetchone()[0]; c.close(); return n def recovery_clear(): c = db(); recovery_table(c); c.execute('DELETE FROM recovery_codes'); c.commit(); c.close() def second_factor_ok(secret, code): return totp_ok(secret, code) or recovery_use(code) def create_user(username, password, secret=None): if not re.fullmatch(r'[A-Za-z0-9_.-]{3,32}', username or ''): raise ValueError('Nom utilisateur invalide') if not 8 <= len(password) <= 128: raise ValueError('Le mot de passe doit contenir entre 8 et 128 caractères') c = db(); c.execute('DELETE FROM auth_users'); c.execute('INSERT INTO auth_users VALUES(?,?,?)', (username, password_hash(password), secret)); c.commit(); c.close() def verify_user(username, password): c = db(); row = c.execute('SELECT username,password_hash,totp_secret FROM auth_users WHERE username=?', (username.strip(),)).fetchone(); c.close() return row if row and password_ok(password, row[1]) else None def new_session(username): token = secrets.token_urlsafe(32); sessions[token] = {'username': username, 'expires': time.time() + SESSION_TTL}; return token def valid_session(h): token = h.headers.get('X-Session-Token', ''); s = sessions.get(token) if isinstance(s, dict) and s['expires'] > time.time(): return True sessions.pop(token, None); return False def username(h): s = sessions.get(h.headers.get('X-Session-Token', ''), {}); return s.get('username', '') if isinstance(s, dict) else '' def command(*args): try: return subprocess.check_output(args, text=True, stderr=subprocess.DEVNULL).strip() except (subprocess.CalledProcessError, FileNotFoundError): return '' def read_json(h): return json.loads(h.rfile.read(int(h.headers.get('Content-Length', '0'))) or b'{}') def request_ip(h): return h.headers.get('X-Real-IP', '').strip() or h.client_address[0] def log(event, name='Nom non communiqué', ip='Non attribuée', mac='—'): try: c = db(); c.execute('INSERT INTO connection_logs(event_time,event,name,ip,mac) VALUES(?,?,?,?,?)', (time.strftime('%Y-%m-%d %H:%M:%S'), event, name, ip, mac)); c.commit(); c.close() except sqlite3.Error: pass def logs(): c = db(); rows = c.execute('SELECT event_time,event,name,ip,mac FROM connection_logs ORDER BY id DESC LIMIT 200').fetchall(); c.close() return [{'time':r[0], 'event':r[1], 'name':r[2], 'ip':r[3], 'mac':r[4]} for r in rows] def connection_name(): active = command('nmcli','-t','-g','GENERAL.CONNECTION','device','show',INTERFACE) if active and active != '--': return active try: return open(CONNECTION_FILE, encoding='utf8').read().strip() except OSError: return os.environ.get('AP_CONNECTION', '') def save_connection(name): try: open(CONNECTION_FILE, 'w', encoding='utf8').write(name) except OSError: pass def security(value): values = {'wpa1':('wpa-psk','wpa','optional'), 'wpa2':('wpa-psk','rsn','optional'), 'wpa3':('sae',None,'required')} value = str(value or 'wpa2').lower() if value not in values: raise ValueError('Sécurité Wi-Fi invalide') key, proto, pmf = values[value]; return value, key, proto, pmf def status(): name = connection_name(); ssid = command('nmcli','-t','-g','802-11-wireless.ssid','connection','show',name) if name else '' password = command('nmcli','--show-secrets','-t','-g','802-11-wireless-security.psk','connection','show',name) if name else '' key = command('nmcli','-t','-g','802-11-wireless-security.key-mgmt','connection','show',name).lower() if name else '' proto = command('nmcli','-t','-g','802-11-wireless-security.proto','connection','show',name).lower() if name else '' sec = 'wpa3' if key == 'sae' else 'wpa1' if proto == 'wpa' else 'wpa2' state = command('nmcli','-t','-g','GENERAL.STATE','device','show',INTERFACE) # CORRECTION : "30 (disconnected)" contient "connected" ; seul le code 100 signifie connecté. active = state.startswith('100') auto = command('nmcli','-t','-g','connection.autoconnect','connection','show',name).lower() == 'yes' if name else False if active and uptime_start() is None: mark_ap_started() return {'ssid':ssid or 'SSID inconnu','password':password,'security':sec,'autoconnect':auto,'interface':INTERFACE,'connection':name,'active':active,'configured':bool(name),'uptime_seconds':ap_uptime() if active else 0} def blocked_set(): c = db(); rows = c.execute('SELECT mac FROM blocked_clients').fetchall(); c.close(); return {r[0] for r in rows} def lease_ip(mac): mac = (mac or '').upper() try: import glob files = ['/var/lib/NetworkManager/dnsmasq.leases', '/var/lib/misc/dnsmasq.leases'] files += glob.glob('/var/lib/NetworkManager/dnsmasq-*.leases') for path in files: with open(path, encoding='utf8') as f: for line in f: fields = line.strip().split() if len(fields) >= 3 and fields[1].upper() == mac: ip = fields[2].strip() if re.fullmatch(r'\d{1,3}(?:\.\d{1,3}){3}', ip): return ip except (OSError, UnicodeError): pass return '' def clients(): neigh = command('ip','neigh','show','dev',INTERFACE); ipmac = {} for line in neigh.splitlines(): m = re.search(r'^(\S+).*?lladdr ([0-9a-f:]{17})', line, re.I) if m: ipmac[m.group(2).upper()] = m.group(1) output = []; dump = command('iw','dev',INTERFACE,'station','dump'); blocked_macs = blocked_set() for m in re.finditer(r'^Station ([0-9a-f:]{17})(.*?)(?=^Station |\Z)', dump, re.I|re.M|re.S): mac = m.group(1).upper(); detail = m.group(2); sm = re.search(r'^\s*signal:\s*(-?\d+)', detail, re.I|re.M); ip = ipmac.get(mac, '') or lease_ip(mac); blocked = mac in blocked_macs; name = device_name(ip or 'Non attribuée', mac) if not ip: try: c = db(); row = c.execute('SELECT last_ip FROM client_names WHERE mac=?', (mac,)).fetchone(); c.close(); ip = row[0] if row and row[0] else 'Non attribuée' except sqlite3.Error: ip = 'Non attribuée' if not name.startswith('Client ') and name != 'Nom non communiqué': try: c = db(); c.execute('INSERT INTO client_names(mac,name,last_ip,last_seen) VALUES(?,?,?,?) ON CONFLICT(mac) DO UPDATE SET name=excluded.name,last_ip=CASE WHEN excluded.last_ip != \'Non attribuée\' THEN excluded.last_ip ELSE client_names.last_ip END,last_seen=excluded.last_seen', (mac,name,ip,time.strftime('%Y-%m-%d %H:%M:%S'))); c.commit(); c.close() except sqlite3.Error: pass output.append({'name':name,'ip':ip,'mac':mac,'signal_dbm':int(sm.group(1)) if sm else None,'blocked':blocked,'connected':not blocked}) return output def known_device(mac): name, ip = '', 'Non attribuée' try: c = db(); row = c.execute('SELECT name,last_ip FROM client_names WHERE mac=?', (mac,)).fetchone(); c.close() if row: name, ip = row[0] or '', row[1] or ip except sqlite3.Error: pass return name or device_name(ip, mac), ip def set_blocked(mac, blocked): mac = mac.upper() if not re.fullmatch(r'[0-9A-F]{2}(:[0-9A-F]{2}){5}', mac): raise ValueError('Adresse MAC invalide') name, ip = known_device(mac) c = db() if blocked: c.execute('INSERT OR IGNORE INTO blocked_clients VALUES(?,?)',(mac,time.strftime('%Y-%m-%d %H:%M:%S'))); command('iw','dev',INTERFACE,'station','del',mac) else: c.execute('DELETE FROM blocked_clients WHERE mac=?',(mac,)) c.commit(); c.close(); log('blocage' if blocked else 'déblocage', name, ip, mac) def blocked_list(): c = db(); rows = c.execute('SELECT b.mac,b.blocked_at,n.name FROM blocked_clients b LEFT JOIN client_names n ON n.mac=b.mac ORDER BY b.blocked_at DESC').fetchall(); c.close() return [{'mac':r[0],'blocked_at':r[1],'name':r[2] or 'Client ' + r[0].replace(':', '')[-4:]} for r in rows] def update_logs(items): global last_macs current = {x['mac']:x for x in items} for mac,x in current.items(): if mac not in last_macs: log('connexion', x['name'], x['ip'], mac) for mac in last_macs - set(current): name, ip = 'Nom non communiqué', 'Non attribuée' try: c = db(); row = c.execute('SELECT name,last_ip FROM client_names WHERE mac=?', (mac,)).fetchone(); c.close() if row: name = row[0] or name ip = row[1] or ip except sqlite3.Error: pass log('déconnexion', name, ip, mac) last_macs = set(current) def configure_ap(data, initial=False, caller_ip=None): ssid, password = str(data.get('ssid','')).strip(), str(data.get('password','')) sec, key, proto, pmf = security(data.get('security','wpa2')) if not ssid or len(ssid)>32 or not 8<=len(password)<=63: raise ValueError('SSID ou mot de passe Wi-Fi invalide') name = ('AP-' + ssid) if initial else connection_name() if not name: raise ValueError('Aucune connexion AP configurée') if initial: r=subprocess.run(['nmcli','connection','add','type','wifi','ifname',INTERFACE,'con-name',name,'ssid',ssid],text=True,capture_output=True) if r.returncode and 'already exists' not in r.stderr.lower(): raise RuntimeError(r.stderr.strip()) args=['nmcli','connection','modify',name,'802-11-wireless.mode','ap','802-11-wireless.ssid',ssid,'802-11-wireless-security.key-mgmt',key,'802-11-wireless-security.psk',password,'802-11-wireless-security.pmf',pmf,'ipv4.method','shared','ipv6.method','ignore','connection.autoconnect','yes' if data.get('autoconnect',True) else 'no'] if proto: args += ['802-11-wireless-security.proto',proto] r=subprocess.run(args,text=True,capture_output=True) if r.returncode: raise RuntimeError(r.stderr.strip() or 'NetworkManager a refusé la configuration') save_connection(name); r=subprocess.run(['nmcli','connection','up',name],text=True,capture_output=True) if r.returncode: raise RuntimeError(r.stderr.strip() or 'Impossible de démarrer le point d’accès') mark_ap_started(); log('création' if initial else 'enregistrement_modifie', 'Réseau Wi-Fi '+ssid, ip=caller_ip or 'Non attribuée', mac='AP'); return name class Handler(BaseHTTPRequestHandler): def send_json(self, data, status=200): body=json.dumps(data,ensure_ascii=False).encode(); self.send_response(status); self.send_header('Content-Type','application/json; charset=utf-8'); self.send_header('Access-Control-Allow-Origin','*'); self.send_header('Access-Control-Allow-Headers','Content-Type, X-Session-Token'); self.send_header('Access-Control-Allow-Methods','GET, POST, OPTIONS'); self.send_header('Content-Length',str(len(body))); self.end_headers(); self.wfile.write(body) def do_OPTIONS(self): self.send_response(204); self.send_header('Access-Control-Allow-Origin','*'); self.send_header('Access-Control-Allow-Headers','Content-Type, X-Session-Token'); self.send_header('Access-Control-Allow-Methods','GET, POST, OPTIONS'); self.end_headers() def do_GET(self): if self.path in ('/','/index.html','/ap_dashboard.html'): try: body=open(os.path.join(os.path.dirname(__file__),'index.html'),'rb').read(); self.send_response(200); self.send_header('Content-Type','text/html; charset=utf-8'); self.send_header('Content-Length',str(len(body))); self.end_headers(); self.wfile.write(body) except OSError: self.send_error(404,'index.html introuvable') return if self.path == '/api/auth/status': self.send_json(auth_state()); return if not valid_session(self): self.send_json({'error':'Authentification requise'},401); return if self.path == '/api/status': self.send_json(status()) elif self.path == '/api/clients': self.send_json(clients()) elif self.path == '/api/clients/blocked': self.send_json(blocked_list()) elif self.path == '/api/logs': self.send_json(logs()) elif self.path == '/api/preferences': c=db(); rows=dict(c.execute("SELECT key,value FROM preferences").fetchall()); c.close() self.send_json({ 'theme': rows.get('theme', 'dark'), 'app_name': rows.get('app_name', 'Administration de l’AP Wi‑Fi'), 'app_short_name': rows.get('app_short_name', 'AP Manageur') }) else: self.send_json({'error':'Route inconnue'},404) def do_POST(self): try: data=read_json(self) except Exception: self.send_json({'error':'JSON invalide'},400); return if not isinstance(data, dict): self.send_json({'error':'JSON invalide'},400); return try: if self.path == '/api/auth/setup': if auth_state()['configured']: raise ValueError('Un compte existe déjà') u=str(data.get('username','')) create_user(u,str(data.get('password','')),None) log('creation_compte_admin', u, request_ip(self), 'AUTH') self.send_json({'ok':True,'message':'Compte créé','two_factor_enabled':False}); return if self.path == '/api/login': u=str(data.get('username','')) row=verify_user(u,str(data.get('password',''))) if not row: log('echec_connexion', u or 'Inconnu', request_ip(self), 'AUTH') raise ValueError('Utilisateur ou mot de passe incorrect') used_recovery = False if row[2]: otp=data.get('otp') if not totp_ok(row[2],otp): if not recovery_use(otp): log('echec_connexion', row[0], request_ip(self), 'AUTH') raise ValueError('Code de double identification incorrect') used_recovery = True log('connexion_code_recuperation', row[0], request_ip(self), 'AUTH') else: log('connexion_admin', row[0], request_ip(self), 'AUTH') else: log('connexion_admin', row[0], request_ip(self), 'AUTH') self.send_json({'ok':True,'token':new_session(row[0]),'used_recovery':used_recovery}); return if self.path == '/api/logout': log('deconnexion_admin', username(self) or 'Administrateur', request_ip(self), 'AUTH') sessions.pop(self.headers.get('X-Session-Token',''),None); self.send_json({'ok':True}); return if not valid_session(self): self.send_json({'error':'Authentification requise'},401); return if self.path == '/api/clients/block': set_blocked(str(data.get('mac','')),bool(data.get('blocked',True))); self.send_json({'ok':True}); return if self.path == '/api/ap/control': action=str(data.get('action','')); name=connection_name() if not name: raise ValueError('Aucune connexion AP configurée') cmd=['nmcli','connection','up' if action=='start' else 'down',name] if action in ('start','stop') else ['nmcli','connection','down',name] r=subprocess.run(cmd,capture_output=True,text=True) if action=='restart': r=subprocess.run(['nmcli','connection','up',name],capture_output=True,text=True) if r.returncode: raise RuntimeError(r.stderr.strip()) if action in ('start','restart'): mark_ap_started() elif action == 'stop': try: os.remove(UPTIME_FILE) except OSError: pass log({'start':'démarrage','stop':'arrêt','restart':'redémarrage'}.get(action,'action'), 'Point d’accès Wi-Fi', ip=request_ip(self), mac='AP'); self.send_json({'ok':True}); return if self.path == '/api/setup': self.send_json({'ok':True,'message':'Configuration enregistrée','connection':configure_ap(data,True,caller_ip=request_ip(self))}); return if self.path == '/api/settings': self.send_json({'ok':True,'message':'Réglages enregistrés','connection':configure_ap(data,caller_ip=request_ip(self))}); return if self.path == '/api/logs/clear': c=db(); c.execute('DELETE FROM connection_logs'); c.commit(); c.close() log('suppression_journaux', username(self) or 'Administrateur', request_ip(self), 'SYSTEM') self.send_json({'ok':True,'message':'Journaux supprimés'}); return if self.path == '/api/data/clear': name = connection_name() names_to_delete = set() if name: names_to_delete.add(name) out = subprocess.run(['nmcli', '-t', '-g', 'NAME', 'connection', 'show'], capture_output=True, text=True).stdout for line in out.splitlines(): line = line.strip() if line.startswith('AP-') or line == name: names_to_delete.add(line) for n in names_to_delete: if n: subprocess.run(['nmcli', 'connection', 'down', n], capture_output=True, text=True) subprocess.run(['nmcli', 'connection', 'delete', n], capture_output=True, text=True) try: os.remove(CONNECTION_FILE) except OSError: pass try: os.remove(UPTIME_FILE) except OSError: pass c=db() c.execute('DELETE FROM connection_logs') c.execute('DELETE FROM blocked_clients') c.execute('DELETE FROM client_names') c.execute('DELETE FROM auth_users') c.execute('DROP TABLE IF EXISTS recovery_codes') c.execute('DELETE FROM preferences') c.commit(); c.close() sessions.clear() log('reinitialisation_donnees', 'Système', request_ip(self), 'SYSTEM') self.send_json({'ok':True,'message':'Toutes les données et le point d’accès ont été réinitialisés et supprimés'}); return if self.path == '/api/auth/2fa/status': self.send_json({'enabled':auth_state()['two_factor'],'recovery_remaining':recovery_remaining()}); return if self.path == '/api/auth/2fa/enable': row=verify_user(username(self),str(data.get('password',''))) if not row: raise ValueError('Mot de passe incorrect') secret=base64.b32encode(secrets.token_bytes(10)).decode('ascii').rstrip('=') sessions[self.headers.get('X-Session-Token','')]['pending_2fa']=secret log('generation_secret_2fa', username(self), request_ip(self), 'AUTH') self.send_json({'ok':True,'message':'Secret généré. Ajoutez-le dans votre application TOTP puis confirmez avec le code affiché.','secret':secret}); return if self.path == '/api/auth/2fa/confirm': session=sessions.get(self.headers.get('X-Session-Token',''),{}) secret=session.get('pending_2fa') if not secret or not totp_ok(secret,data.get('code')): raise ValueError('Code 2FA invalide') row=verify_user(username(self),str(data.get('password',''))) if not row: raise ValueError('Mot de passe incorrect') create_user(row[0],str(data.get('password','')),secret); session.pop('pending_2fa',None) log('activation_2fa', username(self), request_ip(self), 'AUTH') self.send_json({'ok':True,'message':'Double identification activée. Conservez vos codes de récupération.','recovery_codes':recovery_generate()}); return if self.path == '/api/auth/2fa/disable': row=verify_user(username(self),str(data.get('password',''))) if not row: raise ValueError('Mot de passe incorrect') if row[2] and not second_factor_ok(row[2],data.get('code')): raise ValueError('Code 2FA invalide') create_user(row[0],str(data.get('password','')),None); recovery_clear() log('desactivation_2fa', username(self), request_ip(self), 'AUTH') self.send_json({'ok':True,'message':'Double identification désactivée'}); return if self.path == '/api/auth/2fa/recovery': row=verify_user(username(self),str(data.get('password',''))) if not row: raise ValueError('Mot de passe incorrect') if not row[2]: raise ValueError('La 2FA n’est pas activée') if not second_factor_ok(row[2],data.get('code')): raise ValueError('Code 2FA invalide') log('regeneration_codes_secours', username(self), request_ip(self), 'AUTH') self.send_json({'ok':True,'message':'Nouveaux codes générés','recovery_codes':recovery_generate()}); return if self.path == '/api/auth/password': current=verify_user(username(self),str(data.get('current_password',''))) if not current: raise ValueError('Mot de passe actuel incorrect') new_u = str(data.get('username',username(self))) create_user(new_u,str(data.get('password','')),current[2]) log('modification_identifiants', new_u, request_ip(self), 'AUTH') self.send_json({'ok':True,'message':'Identifiants modifiés'}); return if self.path == '/api/preferences': c=db() if 'theme' in data: theme=str(data.get('theme','')) if theme not in ('dark','light','auto'): raise ValueError('Thème invalide') c.execute("INSERT OR REPLACE INTO preferences(key,value) VALUES('theme',?)",(theme,)) if 'app_name' in data: app_name=str(data.get('app_name','')).strip() if not app_name or len(app_name)>64: raise ValueError('Nom de l’application invalide') c.execute("INSERT OR REPLACE INTO preferences(key,value) VALUES('app_name',?)",(app_name,)) if 'app_short_name' in data: app_short_name=str(data.get('app_short_name','')).strip() if not app_short_name or len(app_short_name)>32: raise ValueError('Nom court invalide') c.execute("INSERT OR REPLACE INTO preferences(key,value) VALUES('app_short_name',?)",(app_short_name,)) c.commit(); c.close() log('modification_preferences', username(self) or 'Administrateur', request_ip(self), 'CONFIG') self.send_json({'ok':True,'message':'Préférences enregistrées'}); return self.send_json({'error':'Route inconnue'},404) except (ValueError, RuntimeError, sqlite3.Error) as e: self.send_json({'error':str(e)},400) except Exception: self.send_json({'error':'Erreur interne'},500) def log_message(self,*args): pass def monitor(): while True: try: items = clients() # CORRECTION : éjecter en continu les appareils bloqués qui se reconnectent. for x in items: if x['blocked']: command('iw','dev',INTERFACE,'station','del',x['mac']) update_logs([x for x in items if not x['blocked']]) except Exception: pass time.sleep(2) if __name__ == '__main__': db(); threading.Thread(target=monitor,daemon=True).start(); print(f'API AP sur http://0.0.0.0:{PORT}'); ThreadingHTTPServer((HOST,PORT),Handler).serve_forever()