394 lines
13 KiB
Python
394 lines
13 KiB
Python
import base64
|
|
import os
|
|
import socket
|
|
import time
|
|
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
|
|
from app.core.config import Config
|
|
from app.utils.platform_utils import (
|
|
get_config_dir,
|
|
get_wg_config_dir,
|
|
is_windows,
|
|
run_command,
|
|
run_privileged,
|
|
wg_available,
|
|
)
|
|
|
|
|
|
def generate_keypair() -> tuple[str, str]:
|
|
"""Retourne (private_key_b64, public_key_b64) au format WireGuard."""
|
|
private = X25519PrivateKey.generate()
|
|
private_bytes = private.private_bytes_raw()
|
|
public_bytes = private.public_key().public_bytes_raw()
|
|
return (
|
|
base64.b64encode(private_bytes).decode(),
|
|
base64.b64encode(public_bytes).decode(),
|
|
)
|
|
|
|
|
|
def generate_preshared_key() -> str:
|
|
return base64.b64encode(os.urandom(32)).decode()
|
|
|
|
|
|
def build_client_config(cfg: Config) -> str:
|
|
wg = cfg.wg
|
|
lines = [
|
|
"[Interface]",
|
|
f"PrivateKey = {wg['client_private_key']}",
|
|
f"Address = {wg['client_address']}",
|
|
f"DNS = {wg['dns']}",
|
|
"",
|
|
"[Peer]",
|
|
f"PublicKey = {wg['server_public_key']}",
|
|
f"AllowedIPs = {wg['allowed_ips']}",
|
|
f"Endpoint = {wg['server_endpoint']}:{wg['server_port']}",
|
|
f"PersistentKeepalive = {wg['keepalive']}",
|
|
]
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def get_client_config_path(cfg: Config) -> str:
|
|
name = cfg.wg.get("interface_name", "wgs0")
|
|
if is_windows():
|
|
config_dir = get_config_dir()
|
|
else:
|
|
config_dir = get_wg_config_dir()
|
|
return os.path.join(config_dir, f"{name}.conf")
|
|
|
|
|
|
def write_client_config(cfg: Config) -> tuple[bool, str]:
|
|
content = build_client_config(cfg)
|
|
path = get_client_config_path(cfg)
|
|
try:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w") as f:
|
|
f.write(content)
|
|
if not is_windows():
|
|
os.chmod(path, 0o600)
|
|
return True, path
|
|
except PermissionError:
|
|
# Écriture via sudo sur Linux
|
|
config_dir_path = get_config_dir()
|
|
tmp = os.path.join(config_dir_path, "wgs_tmp.conf")
|
|
with open(tmp, "w") as f:
|
|
f.write(content)
|
|
code, _, err = run_privileged(["cp", tmp, path])
|
|
os.unlink(tmp)
|
|
if code == 0:
|
|
run_privileged(["chmod", "600", path])
|
|
return True, path
|
|
return False, err
|
|
|
|
|
|
def is_connected(cfg: Config) -> bool:
|
|
name = cfg.wg.get("interface_name", "wgs0")
|
|
if is_windows():
|
|
code, out, _ = run_command(["sc", "query", f"WireGuardTunnel${name}"])
|
|
return code == 0 and "RUNNING" in out
|
|
code, out, _ = run_command(["wg", "show", name])
|
|
return code == 0 and bool(out)
|
|
|
|
|
|
def connect(cfg: Config) -> tuple[bool, str]:
|
|
if not cfg.configured:
|
|
return False, "WireGuard non configuré. Ouvrez le panneau Admin."
|
|
|
|
ok, result = write_client_config(cfg)
|
|
if not ok:
|
|
return False, f"Impossible d'écrire la config : {result}"
|
|
|
|
name = cfg.wg.get("interface_name", "wgs0")
|
|
|
|
if is_windows():
|
|
code, _, err = run_command(["wireguard", "/installtunnel", result])
|
|
return (code == 0), (err or "Connecté")
|
|
else:
|
|
code, _, err = run_privileged(["wg-quick", "up", result])
|
|
if code == 0:
|
|
return True, "Tunnel WireGuard activé"
|
|
return False, err or "Erreur lors de la connexion"
|
|
|
|
|
|
def disconnect(cfg: Config) -> tuple[bool, str]:
|
|
name = cfg.wg.get("interface_name", "wgs0")
|
|
|
|
if is_windows():
|
|
code, _, err = run_command(["wireguard", "/uninstalltunnel", name])
|
|
return (code == 0), (err or "Déconnecté")
|
|
else:
|
|
path = get_client_config_path(cfg)
|
|
if not os.path.exists(path):
|
|
path = name
|
|
code, _, err = run_privileged(["wg-quick", "down", path])
|
|
if code == 0:
|
|
return True, "Tunnel WireGuard désactivé"
|
|
return False, err or "Erreur lors de la déconnexion"
|
|
|
|
|
|
def test_connection(cfg: Config, timeout: int = 5) -> tuple[bool, str]:
|
|
endpoint = cfg.wg.get("server_endpoint", "")
|
|
port = int(cfg.wg.get("server_port", 51820))
|
|
if not endpoint:
|
|
return False, "Aucun serveur configuré"
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.settimeout(timeout)
|
|
start = time.monotonic()
|
|
sock.connect((endpoint, port))
|
|
sock.close()
|
|
latency = int((time.monotonic() - start) * 1000)
|
|
return True, f"Serveur joignable ({latency} ms)"
|
|
except socket.timeout:
|
|
return False, "Timeout — serveur inaccessible"
|
|
except OSError as e:
|
|
return False, str(e)
|
|
|
|
|
|
def parse_conf_file(path: str) -> dict | None:
|
|
"""Parse un fichier .conf WireGuard et retourne les valeurs extraites."""
|
|
result: dict = {}
|
|
section = None
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for raw in f:
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith("["):
|
|
section = line.strip("[]").lower()
|
|
continue
|
|
if "=" not in line:
|
|
continue
|
|
key, val = (x.strip() for x in line.split("=", 1))
|
|
if section == "interface":
|
|
if key == "PrivateKey":
|
|
result["client_private_key"] = val
|
|
elif key == "Address":
|
|
result["client_address"] = val.split(",")[0].strip()
|
|
elif key == "DNS":
|
|
result["dns"] = val.split(",")[0].strip()
|
|
elif section == "peer":
|
|
if key == "PublicKey":
|
|
result["server_public_key"] = val
|
|
elif key == "Endpoint" and ":" in val:
|
|
host, port = val.rsplit(":", 1)
|
|
result["server_endpoint"] = host.strip("[]")
|
|
try:
|
|
result["server_port"] = int(port)
|
|
except ValueError:
|
|
pass
|
|
elif key == "AllowedIPs":
|
|
result["allowed_ips"] = val
|
|
elif key == "PersistentKeepalive":
|
|
try:
|
|
result["keepalive"] = int(val)
|
|
except ValueError:
|
|
pass
|
|
# Dériver la clé publique depuis la clé privée importée
|
|
if "client_private_key" in result and "client_public_key" not in result:
|
|
try:
|
|
priv_bytes = base64.b64decode(result["client_private_key"])
|
|
priv = X25519PrivateKey.from_private_bytes(priv_bytes)
|
|
pub_bytes = priv.public_key().public_bytes_raw()
|
|
result["client_public_key"] = base64.b64encode(pub_bytes).decode()
|
|
except Exception:
|
|
pass
|
|
return result or None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def export_conf_file(cfg: Config, dest_path: str) -> tuple[bool, str]:
|
|
"""Exporte la configuration courante vers un fichier .conf."""
|
|
try:
|
|
content = build_client_config(cfg)
|
|
with open(dest_path, "w") as f:
|
|
f.write(content)
|
|
if not is_windows():
|
|
os.chmod(dest_path, 0o600)
|
|
return True, dest_path
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
|
|
def get_interface_bytes(iface: str) -> tuple[int, int] | None:
|
|
"""Retourne (rx_bytes, tx_bytes) depuis /proc/net/dev (Linux)."""
|
|
try:
|
|
with open("/proc/net/dev") as f:
|
|
for line in f:
|
|
if iface in line:
|
|
parts = line.split()
|
|
return int(parts[1]), int(parts[9])
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def ping_server(host: str, timeout: int = 2) -> int | None:
|
|
"""Ping ICMP du serveur. Retourne la latence en ms, ou None si inaccessible."""
|
|
if not host:
|
|
return None
|
|
if is_windows():
|
|
code, out, _ = run_command(
|
|
["ping", "-n", "1", "-w", str(timeout * 1000), host], timeout + 2
|
|
)
|
|
else:
|
|
code, out, _ = run_command(
|
|
["ping", "-c", "1", "-W", str(timeout), host], timeout + 2
|
|
)
|
|
if code != 0:
|
|
return None
|
|
# Parser "time=X.X ms" ou "temps=X.X ms"
|
|
import re
|
|
m = re.search(r"[Tt]ime[<=]([\d.]+)\s*ms", out)
|
|
if m:
|
|
return int(float(m.group(1)))
|
|
return None
|
|
|
|
|
|
# ── Test DNS leak ─────────────────────────────────────────────────────────
|
|
|
|
def dns_leak_test(cfg: Config) -> dict:
|
|
"""
|
|
Teste si le DNS fuit en dehors du tunnel.
|
|
Retourne {"status": ok|leak|unknown, "resolvers": [...], "expected": str}
|
|
"""
|
|
expected_dns = cfg.wg.get("dns", "")
|
|
resolvers: list[str] = []
|
|
|
|
# 1. Lire /etc/resolv.conf (Linux) ou ipconfig /all (Windows)
|
|
if not is_windows():
|
|
try:
|
|
with open("/etc/resolv.conf") as f:
|
|
for line in f:
|
|
if line.startswith("nameserver"):
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
resolvers.append(parts[1])
|
|
except Exception:
|
|
pass
|
|
else:
|
|
code, out, _ = run_command(["ipconfig", "/all"])
|
|
if code == 0:
|
|
import re
|
|
resolvers = re.findall(r"DNS Servers.*?:(.*?)(?:\n\S|\Z)",
|
|
out, re.DOTALL)
|
|
|
|
# 2. Résoudre un domaine test via dig/nslookup pour voir quel serveur répond
|
|
probe_ip: str = ""
|
|
code, out, _ = run_command(["dig", "+short", "+time=3", "whoami.akamai.net"], 6)
|
|
if code == 0:
|
|
lines = [l.strip() for l in out.splitlines() if l.strip()]
|
|
if lines:
|
|
probe_ip = lines[-1]
|
|
|
|
if not resolvers:
|
|
return {"status": "unknown", "resolvers": [], "expected": expected_dns,
|
|
"probe_ip": probe_ip}
|
|
|
|
# 3. Comparer avec le DNS configuré
|
|
if expected_dns and any(expected_dns in r for r in resolvers):
|
|
status = "ok"
|
|
elif expected_dns:
|
|
status = "leak"
|
|
else:
|
|
status = "unknown"
|
|
|
|
return {
|
|
"status": status,
|
|
"resolvers": resolvers,
|
|
"expected": expected_dns,
|
|
"probe_ip": probe_ip,
|
|
}
|
|
|
|
|
|
# ── Génération config serveur ─────────────────────────────────────────────
|
|
|
|
def generate_server_config(
|
|
server_port: int = 51820,
|
|
server_address: str = "10.8.0.1/24",
|
|
client_address: str = "10.8.0.2/24",
|
|
client_allowed_ips: str = "10.8.0.2/32",
|
|
) -> dict:
|
|
"""
|
|
Génère une paire complète (server_conf, client_conf) avec de nouvelles clés.
|
|
Retourne un dict avec server_priv/pub, client_priv/pub, server_conf, client_conf.
|
|
"""
|
|
srv_priv, srv_pub = generate_keypair()
|
|
cli_priv, cli_pub = generate_keypair()
|
|
psk = generate_preshared_key()
|
|
|
|
server_conf = "\n".join([
|
|
"[Interface]",
|
|
f"PrivateKey = {srv_priv}",
|
|
f"Address = {server_address}",
|
|
f"ListenPort = {server_port}",
|
|
"PostUp = iptables -A FORWARD -i %i -j ACCEPT; iptables -t nat -A POSTROUTING -o eth0 -j MASQUERADE",
|
|
"PostDown = iptables -D FORWARD -i %i -j ACCEPT; iptables -t nat -D POSTROUTING -o eth0 -j MASQUERADE",
|
|
"",
|
|
"# === Peer client ===",
|
|
"[Peer]",
|
|
f"PublicKey = {cli_pub}",
|
|
f"PresharedKey = {psk}",
|
|
f"AllowedIPs = {client_allowed_ips}",
|
|
])
|
|
|
|
client_conf = "\n".join([
|
|
"[Interface]",
|
|
f"PrivateKey = {cli_priv}",
|
|
f"Address = {client_address}",
|
|
"DNS = 1.1.1.1",
|
|
"",
|
|
"[Peer]",
|
|
f"PublicKey = {srv_pub}",
|
|
f"PresharedKey = {psk}",
|
|
"AllowedIPs = 0.0.0.0/0",
|
|
"Endpoint = <SERVER_IP>:" + str(server_port),
|
|
"PersistentKeepalive = 25",
|
|
])
|
|
|
|
return {
|
|
"server_priv": srv_priv, "server_pub": srv_pub,
|
|
"client_priv": cli_priv, "client_pub": cli_pub,
|
|
"psk": psk,
|
|
"server_conf": server_conf,
|
|
"client_conf": client_conf,
|
|
}
|
|
|
|
|
|
def get_status_info(cfg: Config) -> dict:
|
|
name = cfg.wg.get("interface_name", "wgs0")
|
|
info = {
|
|
"connected": False,
|
|
"interface": name,
|
|
"peer": "",
|
|
"rx_bytes": 0,
|
|
"tx_bytes": 0,
|
|
"last_handshake": "",
|
|
}
|
|
if is_windows():
|
|
info["connected"] = is_connected(cfg)
|
|
return info
|
|
|
|
code, out, _ = run_command(["wg", "show", name])
|
|
if code != 0 or not out:
|
|
return info
|
|
|
|
info["connected"] = True
|
|
for line in out.splitlines():
|
|
line = line.strip()
|
|
if line.startswith("peer:"):
|
|
info["peer"] = line.split(":", 1)[1].strip()[:16] + "…"
|
|
elif line.startswith("transfer:"):
|
|
parts = line.split(":", 1)[1].strip().split(",")
|
|
try:
|
|
rx = parts[0].strip().split()[0]
|
|
tx = parts[1].strip().split()[0]
|
|
info["rx_bytes"] = rx
|
|
info["tx_bytes"] = tx
|
|
except Exception:
|
|
pass
|
|
elif line.startswith("latest handshake:"):
|
|
info["last_handshake"] = line.split(":", 1)[1].strip()
|
|
|
|
return info
|