506 lines
30 KiB
Python
506 lines
30 KiB
Python
#!/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()
|