Sync from GitHub with Git LFS

agents/peer_sync.py

@@ -8,9 +8,10 @@ import uuid
 import ipaddress
 import re
 import netifaces
+import select
+
 from tools.storage import Storage
 from datetime import datetime
-import select
 
 storage = Storage()
 my_id = storage.get_config_value("agent_id", str(uuid.uuid4()))
@@ -126,147 +127,107 @@ def expand_and_filter(addresses):
     return out
 
 # ======================
-# IPv4 UDP LAN Discovery (приём + рассылка по /24)
+# UDP LAN Discovery (IPv4 broadcast + IPv6 multicast)
 # ======================
-def udp_lan_discovery_v4():
+def udp_lan_discovery():
     DISCOVERY_INTERVAL = 30
+
     local_addresses = storage.get_config_value("local_addresses", [])
-    # порты, на которые слать/listen
+
+    # Получаем уникальные порты из local_addresses
     udp_port_set = set()
     for a in local_addresses:
-        try:
-            _proto, hostport = a.split("://", 1)
-            _h, p = parse_hostport(hostport)
-            if p:
-                udp_port_set.add(p)
-        except:
-            continue
+        _, port = parse_hostport(a.split("://", 1)[1])
+        if port:
+            udp_port_set.add(port)
 
-    # слушатели IPv4
+    # ------------------ Слушатели ------------------
     listen_sockets = []
     for port in udp_port_set:
+        # IPv4
         try:
             sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             sock.bind(("", port))
             listen_sockets.append(sock)
-            print(f"[UDPv4] Слушаем UDP на 0.0.0.0:{port}")
+            print(f"[UDP/LAN Discovery] слушаем IPv4 на *:{port}")
         except Exception as e:
-            print(f"[UDPv4] Не удалось слушать порт {port}: {e}")
-
-    send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    send_sock.settimeout(0.5)
+            print(f"[UDP/LAN Discovery] Не удалось создать IPv4 сокет {port}: {e}")
 
-    while True:
-        # --- приём ---
-        for sock in listen_sockets:
-            try:
-                sock.settimeout(0.05)
-                data, addr = sock.recvfrom(4096)
-                msg = json.loads(data.decode("utf-8"))
-                peer_id = msg.get("id")
-                if not peer_id or peer_id == my_id:
-                    continue
-                name = msg.get("name", "unknown")
-                addresses = msg.get("addresses", [])
-                expanded = expand_and_filter(addresses)
-                if expanded:
-                    print(f"[UDPv4] RX от {addr} -> {expanded}")
-                    storage.add_or_update_peer(peer_id, name, expanded, "discovery", "online")
-            except socket.timeout:
-                pass
-            except Exception as e:
-                print(f"[UDPv4] Ошибка при обработке пакета: {e}")
-
-        # --- рассылка ---
-        v4s = get_all_local_ips_v4()
-        msg_data = json.dumps({"id": my_id, "name": agent_name, "addresses": local_addresses}).encode("utf-8")
-        for local_ip in v4s:
-            try:
-                net = ipaddress.ip_network(local_ip + "/24", strict=False)
-            except Exception:
-                continue
-            for ip in net.hosts():
-                ip_str = str(ip)
-                if ip_str == local_ip or ip_str.startswith("127.") or ip_str.startswith("0."):
-                    continue
-                for port in udp_port_set:
-                    try:
-                        send_sock.sendto(msg_data, (ip_str, port))
-                    except:
-                        pass
-
-        time.sleep(DISCOVERY_INTERVAL)
-
-# ======================
-# IPv6 UDP Discovery (приём + мультикаст ff02::1)
-# ======================
-def udp_lan_discovery_v6():
-    DISCOVERY_INTERVAL = 30
-    local_addresses = storage.get_config_value("local_addresses", [])
-    udp_port_set = set()
-    for a in local_addresses:
-        try:
-            _proto, hostport = a.split("://", 1)
-            h, p = parse_hostport(hostport)
-            if p:
-                udp_port_set.add(p)
-        except:
-            continue
-
-    # слушатели IPv6
-    listen_sockets = []
-    for port in udp_port_set:
+        # IPv6
         try:
             sock6 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
             sock6.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             sock6.bind(("::", port))
             listen_sockets.append(sock6)
-            print(f"[UDPv6] Слушаем UDP на [::]:{port}")
+            print(f"[UDP/LAN Discovery] слушаем IPv6 на [::]:{port}")
         except Exception as e:
-            print(f"[UDPv6] Не удалось слушать порт {port}: {e}")
-
-    # отправитель IPv6 (мультикаст link-local all-nodes)
-    try:
-        send_sock6 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
-        # hop limit 1
-        send_sock6.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 1)
-    except Exception as e:
-        print(f"[UDPv6] Не удалось создать отправитель: {e}")
-        send_sock6 = None
+            print(f"[UDP/LAN Discovery] Не удалось создать IPv6 сокет {port}: {e}")
 
-    maddr = "ff02::1"  # all-nodes on-link
-    msg_data = json.dumps({"id": my_id, "name": agent_name, "addresses": local_addresses}).encode("utf-8")
+    msg_data = json.dumps({
+        "id": my_id,
+        "name": agent_name,
+        "addresses": local_addresses
+    }).encode("utf-8")
 
     while True:
-        # --- приём ---
-        for sock in listen_sockets:
+        # ------------------ Приём ------------------
+        rlist, _, _ = select.select(listen_sockets, [], [], 0.5)
+        for sock in rlist:
             try:
-                sock.settimeout(0.05)
-                data, addr = sock.recvfrom(8192)
+                data, addr = sock.recvfrom(2048)
                 msg = json.loads(data.decode("utf-8"))
                 peer_id = msg.get("id")
-                if not peer_id or peer_id == my_id:
+                if peer_id == my_id:
                     continue
+
                 name = msg.get("name", "unknown")
                 addresses = msg.get("addresses", [])
-                expanded = expand_and_filter(addresses)
-                if expanded:
-                    print(f"[UDPv6] RX от {addr} -> {expanded}")
-                    storage.add_or_update_peer(peer_id, name, expanded, "discovery", "online")
-            except socket.timeout:
-                pass
+                storage.add_or_update_peer(peer_id, name, addresses, "discovery", "online")
+                print(f"[UDP/LAN Discovery] peer={peer_id} name={name} addresses={addresses} from {addr}")
             except Exception as e:
-                print(f"[UDPv6] Ошибка при обработке пакета: {e}")
+                print(f"[UDP/LAN Discovery] ошибка при приёме: {e}")
+
+        # ------------------ Отправка ------------------
+        for port in udp_port_set:
+            # ---------------- IPv4 Broadcast ----------------
+            for iface in netifaces.interfaces():
+                addrs = netifaces.ifaddresses(iface).get(netifaces.AF_INET, [])
+                for a in addrs:
+                    if "broadcast" in a:
+                        bcast = a["broadcast"]
+                        try:
+                            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                            sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+                            sock.sendto(msg_data, (bcast, port))
+                            sock.close()
+                            print(f"[UDP/LAN Discovery] -> IPv4 broadcast {bcast}:{port}")
+                        except Exception as e:
+                            print(f"[UDP/LAN Discovery] ошибка IPv4 broadcast {bcast}:{port}: {e}")
+
+            # ---------------- IPv6 Multicast ----------------
+            for iface in netifaces.interfaces():
+                ifaddrs = netifaces.ifaddresses(iface).get(netifaces.AF_INET6, [])
+                for a in ifaddrs:
+                    addr = a.get("addr")
+                    if not addr:
+                        continue
 
-        # --- мультикаст рассылка ---
-        if send_sock6:
-            for port in udp_port_set:
-                try:
-                    # iface scope index не задаём — ОС выберет подходящий on-link интерфейс
-                    send_sock6.sendto(msg_data, (maddr, port))
-                except:
-                    pass
+                    # Если link-local, добавляем scope_id
+                    if addr.startswith("fe80:"):
+                        multicast_addr = f"ff02::1%{iface}"
+                    else:
+                        multicast_addr = "ff02::1"
+
+                    try:
+                        sock6 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
+                        sock6.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, socket.if_nametoindex(iface))
+                        sock6.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 1)
+                        sock6.sendto(msg_data, (multicast_addr, port))
+                        sock6.close()
+                        print(f"[UDP/LAN Discovery] -> IPv6 multicast {multicast_addr}:{port}")
+                    except Exception:
+                        continue
 
         time.sleep(DISCOVERY_INTERVAL)
 
@@ -322,9 +283,9 @@ def udp_discovery_sender():
 
         time.sleep(1)
 
-# ======================
-# TCP Listener (v4+v6) для Peer Exchange
-# ======================
+# ==========================
+# TCP Listener (IPv4 + IPv6, link-local + global)
+# ==========================
 def tcp_listener():
     sockets = []
     for host, port in tcp_ports:
@@ -352,106 +313,110 @@ def tcp_listener():
         if not sockets:
             time.sleep(1)
             continue
+
         readable, _, _ = select.select(sockets, [], [], 1)
         for s in readable:
             try:
                 conn, addr = s.accept()
                 data = b""
                 try:
-                    chunk = conn.recv(1024)
-                    data = chunk or b""
+                    data = conn.recv(1024) or b""
                 except:
-                    data = b""
+                    pass
 
                 if data == b"PEER_EXCHANGE_REQUEST":
                     print(f"[TCP] PEER_EXCHANGE_REQUEST от {addr}")
                     try:
-                        peers = []
+                        peers_list = []
                         for pid, addresses_json in storage.get_online_peers(limit=50):
                             try:
                                 addresses = json.loads(addresses_json)
-                            except Exception:
+                            except:
                                 addresses = []
-                            # фильтруем/расширяем перед отдачей
                             addresses = expand_and_filter(addresses)
-                            peers.append({"id": pid, "addresses": addresses})
-                        payload = json.dumps(peers).encode("utf-8")
+                            peers_list.append({"id": pid, "addresses": addresses})
+                        payload = json.dumps(peers_list).encode("utf-8")
                         conn.sendall(payload)
-                        print(f"[TCP] Отправлен список пиров ({len(peers)}) в {addr}")
+                        print(f"[TCP] Отправлен список пиров ({len(peers_list)}) в {addr}")
                     except Exception as e:
                         print(f"[TCP] Ошибка при отправке списка пиров: {e}")
                 conn.close()
             except Exception as e:
                 print(f"[TCP] Ошибка при обработке соединения: {e}")
 
-# ======================
-# Peer Exchange (TCP инициатор, читает ответ и пишет в БД)
-# ======================
+# ==========================
+# Peer Exchange (инициатор TCP)
+# ==========================
 def peer_exchange():
-    PEER_EXCHANGE_INTERVAL = 30  # почаще, чтобы быстрее раскидать пиров
+    PEER_EXCHANGE_INTERVAL = 120
     while True:
-        rows = storage.get_online_peers(limit=50)
-        for pid, addresses_json in rows:
-            if pid == my_id:
+        peers = storage.get_online_peers(limit=50)
+        for peer in peers:
+            peer_id, addresses_json = peer["id"], peer["addresses"]
+            if peer_id == my_id:
                 continue
+
             try:
                 addr_list = json.loads(addresses_json)
             except:
                 addr_list = []
+
             for addr in addr_list:
                 norm = storage.normalize_address(addr)
                 if not norm:
                     continue
-                proto, hostport = norm.split("://", 1)
+
+                proto, hostport = norm.split("://")
                 if proto not in ["tcp", "any"]:
                     continue
+
                 host, port = parse_hostport(hostport)
-                if not host or not port or is_loopback(host):
+                if not host or not port:
                     continue
 
                 try:
-                    family = socket.AF_INET6 if is_ipv6(host) else socket.AF_INET
-                    s = socket.socket(family, socket.SOCK_STREAM)
-                    s.settimeout(3)
-                    # IPv6 connect tuple принимает (host, port, flowinfo, scope_id), но flow/scope 0 по умолчанию ок
-                    s.connect((host, port))
-                    s.sendall(b"PEER_EXCHANGE_REQUEST")
-
-                    # читаем весь ответ (до закрытия сокета/таймаута)
-                    chunks = []
-                    try:
-                        while True:
-                            b = s.recv(8192)
-                            if not b:
+                    # Для link-local IPv6 автоматически добавляем scope_id
+                    if is_ipv6(host) and host.startswith("fe80:"):
+                        for iface in get_all_local_ips_v6():  # перебор интерфейсов
+                            if iface.endswith(host):  # простой вариант сопоставления
+                                scope_id = socket.if_nametoindex(iface)
+                                sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+                                sock.settimeout(3)
+                                sock.connect((host, port, 0, scope_id))
                                 break
-                            chunks.append(b)
-                    except socket.timeout:
-                        pass
-                    s.close()
+                        else:
+                            continue
+                    else:
+                        sock = socket.socket(socket.AF_INET6 if is_ipv6(host) else socket.AF_INET, socket.SOCK_STREAM)
+                        sock.settimeout(3)
+                        sock.connect((host, port))
 
-                    if not chunks:
+                    sock.sendall(b"PEER_EXCHANGE_REQUEST")
+                    data = sock.recv(64 * 1024)
+                    sock.close()
+
+                    if not data:
                         continue
 
                     try:
-                        peers_list = json.loads(b"".join(chunks).decode("utf-8"))
+                        peers_recv = json.loads(data.decode("utf-8"))
+                        for p in peers_recv:
+                            if p.get("id") and p["id"] != my_id:
+                                storage.add_or_update_peer(
+                                    p["id"],
+                                    p.get("name", "unknown"),
+                                    p.get("addresses", []),
+                                    "peer_exchange",
+                                    "online"
+                                )
+                        print(f"[PeerExchange] Получено {len(peers_recv)} пиров от {host}:{port}")
                     except Exception as e:
-                        print(f"[PeerExchange] ошибка парсинга ответа: {e}")
-                        continue
+                        print(f"[PeerExchange] Ошибка разбора ответа от {host}:{port} -> {e}")
+                    break  # успешное соединение, не пробуем остальные адреса этого пира
 
-                    for p in peers_list:
-                        p_id = p.get("id")
-                        if not p_id or p_id == my_id:
-                            continue
-                        addrs = expand_and_filter(p.get("addresses", []))
-                        if addrs:
-                            storage.add_or_update_peer(p_id, "unknown", addrs, "exchange", "online")
-                            print(f"[PeerExchange] получили {p_id} -> {addrs}")
-
-                    # если успешно отработали по одному TCP-адресу — дальше по этому пиру можно не перебирать
-                    break
-                except Exception as e:
-                    # пробуем следующий адрес пира
+                except Exception:
                     continue
+
         time.sleep(PEER_EXCHANGE_INTERVAL)
 
 # ======================
@@ -461,11 +426,10 @@ def start_sync():
     print("[PeerSync] Запуск фоновой синхронизации")
     storage.load_bootstrap()
 
-    threading.Thread(target=udp_lan_discovery_v4, daemon=True).start()
-    threading.Thread(target=udp_lan_discovery_v6, daemon=True).start()
+    threading.Thread(target=udp_lan_discovery, daemon=True).start()
     threading.Thread(target=udp_discovery_sender, daemon=True).start()
     threading.Thread(target=peer_exchange, daemon=True).start()
     threading.Thread(target=tcp_listener, daemon=True).start()
 
     while True:
-        time.sleep(60)
+        time.sleep(60)