Sync from GitHub with Git LFS

agents/peer_sync.py

@@ -7,58 +7,27 @@ import threading
 import select
 import netifaces
 import re
+import ipaddress
 
 from datetime import datetime, timezone as UTC
 from tools.storage import Storage
 
 storage = Storage()
 
-# ---------------------------
-# Вспомогательные функции
-# ---------------------------
-def parse_hostport(s: str):
-    """
-    Разбирает "IP:port" или "[IPv6]:port" и возвращает (host, port)
-    """
-    s = s.strip()
-    if s.startswith("["):
-        # IPv6 с портом: [addr]:port
-        host, _, port = s[1:].partition("]:")
-        try:
-            port = int(port)
-        except:
-            port = None
-        return host, port
-    else:
-        # IPv4 или IPv6 без []
-        if ":" in s:
-            host, port = s.rsplit(":", 1)
-            try:
-                port = int(port)
-            except:
-                port = None
-            return host, port
-        return s, None
-
-def is_ipv6(host: str):
-    try:
-        socket.inet_pton(socket.AF_INET6, host)
-        return True
-    except OSError:
-        return False
-
 # ---------------------------
 # Конфигурация
 # ---------------------------
 my_id = storage.get_config_value("agent_id")
 agent_name = storage.get_config_value("agent_name", "unknown")
 local_addresses = storage.get_config_value("local_addresses", [])
+global_addresses = storage.get_config_value("global_addresses", [])
+all_addresses = local_addresses + global_addresses  # один раз
 
-# Получаем уникальные локальные порты для прослушки TCP/UDP
+# Получаем уникальные локальные порты
 def get_local_ports():
     ports = set()
     for addr in local_addresses:
-        _, port = parse_hostport(addr.split("://", 1)[1])
+        _, port = storage.parse_hostport(addr.split("://", 1)[1])
         if port:
             ports.add(port)
     return sorted(ports)
@@ -199,19 +168,17 @@ def udp_discovery():
         time.sleep(DISCOVERY_INTERVAL)
 
 # ---------------------------
-# TCP Peer Exchange
+# TCP Peer Exchange (исходящие)
 # ---------------------------
 def tcp_peer_exchange():
-    PEER_EXCHANGE_INTERVAL = 20  # для отладки сделаем меньше
+    PEER_EXCHANGE_INTERVAL = 20  # для отладки
     while True:
         peers = storage.get_known_peers(my_id, limit=50)
         print(f"[PeerExchange] Checking {len(peers)} peers (raw DB)...")
 
         for peer in peers:
-            if isinstance(peer, dict):
-                peer_id, addresses_json = peer["id"], peer["addresses"]
-            else:
-                peer_id, addresses_json = peer[0], peer[1]
+            peer_id = peer["id"] if isinstance(peer, dict) else peer[0]
+            addresses_json = peer["addresses"] if isinstance(peer, dict) else peer[1]
 
             if peer_id == my_id:
                 continue
@@ -222,49 +189,46 @@ def tcp_peer_exchange():
                 print(f"[PeerExchange] JSON decode error for peer {peer_id}: {e}")
                 addr_list = []
 
-            print(f"[PeerExchange] Peer {peer_id} -> addresses={addr_list}")
-
             for addr in addr_list:
                 norm = storage.normalize_address(addr)
                 if not norm:
                     continue
-
                 proto, hostport = norm.split("://", 1)
                 if proto not in ["tcp", "any"]:
                     continue
-
-                host, port = parse_hostport(hostport)
+                host, port = storage.parse_hostport(hostport)
                 if not host or not port:
                     continue
 
                 print(f"[PeerExchange] Trying {peer_id} at {host}:{port} (proto={proto})")
-
                 try:
                     # IPv6 link-local
-                    if is_ipv6(host) and host.startswith("fe80:"):
-                        scope_id = None
-                        for iface in netifaces.interfaces():
-                            if iface.endswith(host):
-                                scope_id = socket.if_nametoindex(iface)
-                                break
-                        if scope_id is not None:
-                            sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
-                            sock.settimeout(3)
-                            sock.connect((host, port, 0, scope_id))
-                        else:
-                            print(f"[PeerExchange] Skipping {host}, no scope_id found")
+                    if storage.is_ipv6(host) and host.startswith("fe80:"):
+                        scope_id = storage.get_ipv6_scope(host)
+                        if scope_id is None:
+                            print(f"[PeerExchange] Skipping {host}, no scope_id")
                             continue
+                        sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+                        sock.settimeout(3)
+                        sock.connect((host, port, 0, scope_id))
                     else:
-                        sock = socket.socket(socket.AF_INET6 if is_ipv6(host) else socket.AF_INET, socket.SOCK_STREAM)
+                        sock = socket.socket(socket.AF_INET6 if storage.is_ipv6(host) else socket.AF_INET,
+                                             socket.SOCK_STREAM)
                         sock.settimeout(3)
                         sock.connect((host, port))
 
-                    # 🔑 Отправляем данные о себе
+                    # LAN или Интернет
+                    if storage.is_private(host):
+                        send_addresses = all_addresses
+                    else:
+                        send_addresses = [a for a in all_addresses
+                                          if is_public(stprage.parse_hostport(a.split("://", 1)[1])[0])]
+
                     handshake = {
                         "type": "PEER_EXCHANGE_REQUEST",
                         "id": my_id,
                         "name": agent_name,
-                        "addresses": my_addresses,  # список вроде ["tcp4://192.168.0.10:4010"]
+                        "addresses": send_addresses,
                     }
                     sock.sendall(json.dumps(handshake).encode("utf-8"))
 
@@ -288,7 +252,7 @@ def tcp_peer_exchange():
                         print(f"[PeerExchange] Decode error from {host}:{port} -> {e}")
                         continue
 
-                    break  # успешное соединение — идём к следующему пиру
+                    break
                 except Exception as e:
                     print(f"[PeerExchange] Connection to {host}:{port} failed: {e}")
                     continue
@@ -296,52 +260,41 @@ def tcp_peer_exchange():
         time.sleep(PEER_EXCHANGE_INTERVAL)
 
 # ---------------------------
-# TCP Listener (обработка входящих PEER_EXCHANGE_REQUEST)
+# TCP Listener (входящие)
 # ---------------------------
 def tcp_listener():
     listen_sockets = []
-
-    # Создаём TCP сокеты на всех локальных портах
     for port in local_ports:
-        # IPv4
-        try:
-            sock4 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            sock4.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            sock4.bind(("", port))
-            sock4.listen(5)
-            listen_sockets.append(sock4)
-            print(f"[TCP Listener] Listening IPv4 on *:{port}")
-        except Exception as e:
-            print(f"[TCP Listener] IPv4 bind failed on port {port}: {e}")
-
-        # IPv6
-        try:
-            sock6 = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
-            sock6.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            sock6.bind(("::", port))
-            sock6.listen(5)
-            listen_sockets.append(sock6)
-            print(f"[TCP Listener] Listening IPv6 on [::]:{port}")
-        except Exception as e:
-            print(f"[TCP Listener] IPv6 bind failed on port {port}: {e}")
+        for family, addr_str in [(socket.AF_INET, ""), (socket.AF_INET6, "::")]:
+            try:
+                sock = socket.socket(family, socket.SOCK_STREAM)
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                sock.bind((addr_str, port))
+                sock.listen(5)
+                listen_sockets.append(sock)
+                proto_str = "IPv6" if family == socket.AF_INET6 else "IPv4"
+                print(f"[TCP Listener] Listening {proto_str} on {addr_str}:{port}")
+            except Exception as e:
+                print(f"[TCP Listener] {proto_str} bind failed on port {port}: {e}")
 
     while True:
         if not listen_sockets:
             time.sleep(1)
             continue
+
         rlist, _, _ = select.select(listen_sockets, [], [], 1)
         for s in rlist:
             try:
                 conn, addr = s.accept()
-                data = conn.recv(1024)
-
+                data = conn.recv(64 * 1024)
                 if not data:
                     conn.close()
                     continue
 
                 try:
                     msg = json.loads(data.decode("utf-8"))
-                except Exception:
+                except Exception as e:
+                    print(f"[TCP Listener] JSON decode error from {addr}: {e}")
                     conn.close()
                     continue
 
@@ -350,48 +303,41 @@ def tcp_listener():
                     peer_name = msg.get("name", "unknown")
                     peer_addrs = msg.get("addresses", [])
 
-                    # Сохраняем нового пира
-                    storage.add_or_update_peer(
-                        peer_id,
-                        peer_name,
-                        peer_addrs,
-                        source="incoming",
-                        status="online"
-                    )
+                    storage.add_or_update_peer(peer_id, peer_name, peer_addrs,
+                                               source="incoming", status="online")
                     print(f"[TCP Listener] Handshake from {peer_id} ({addr})")
 
-                    # Отправляем список известных пиров
+                    # LAN или Интернет
+                    is_lan = storage.is_private(addr[0])
+
                     peers_list = []
                     for peer in storage.get_known_peers(my_id, limit=50):
                         peer_id = peer["id"]
-                        addresses_json = peer["addresses"]
                         try:
-                            addresses = json.loads(addresses_json)
+                            addresses = json.loads(peer["addresses"])
                         except:
                             addresses = []
 
-                        # Обработка IPv6 link-local: добавить scope_id
                         updated_addresses = []
                         for a in addresses:
                             proto, hostport = a.split("://")
-                            host, port = parse_hostport(hostport)
-                            if is_ipv6(host) and host.startswith("fe80:"):
-                                scope_id = None
-                                for iface in netifaces.interfaces():
-                                    iface_addrs = netifaces.ifaddresses(iface).get(netifaces.AF_INET6, [])
-                                    for addr_info in iface_addrs:
-                                        if addr_info.get("addr") == host:
-                                            scope_id = socket.if_nametoindex(iface)
-                                            break
-                                    if scope_id:
-                                        break
+                            host, port = storage.parse_hostport(hostport)
+
+                            # Фильтруем по LAN/Internet
+                            if not is_lan and not is_public(host):
+                                continue
+
+                            # IPv6 link-local
+                            if storage.is_ipv6(host) and host.startswith("fe80:"):
+                                scope_id = storage.get_ipv6_scope(host)
                                 if scope_id:
                                     host = f"{host}%{scope_id}"
+
                             updated_addresses.append(f"{proto}://{host}:{port}")
+
                         peers_list.append({"id": peer_id, "addresses": updated_addresses})
 
-                    payload = json.dumps(peers_list).encode("utf-8")
-                    conn.sendall(payload)
+                    conn.sendall(json.dumps(peers_list).encode("utf-8"))
 
                 conn.close()
             except Exception as e:
@@ -401,13 +347,9 @@ def tcp_listener():
 # Запуск потоков
 # ---------------------------
 def start_sync(bootstrap_file="bootstrap.txt"):
-    # Загружаем bootstrap-пиров перед запуском discovery/peer exchange
     load_bootstrap_peers(bootstrap_file)
-
-    # Печать локальных портов для логов
     print(f"[PeerSync] Local ports: {local_ports}")
 
-    # Запуск потоков
     threading.Thread(target=udp_discovery, daemon=True).start()
     threading.Thread(target=tcp_peer_exchange, daemon=True).start()
     threading.Thread(target=tcp_listener, daemon=True).start()

agents/tools/storage.py

@@ -6,8 +6,11 @@ import os
 import json
 import uuid
 import time
+import socket
+import ipaddress
+import netifaces
 
-from datetime import datetime, timedelta, UTC, timezone
+from datetime import datetime, timedelta, UTC, timezone, timezone as UTC
 from werkzeug.security import generate_password_hash, check_password_hash
 from tools.identity import generate_did
 from tools.crypto import generate_keypair
@@ -17,6 +20,7 @@ UTC = timezone.utc
 SCRIPTS_BASE_PATH = "scripts"
 
 class Storage:
+    _scope_cache = {}
     def __init__(self, config=None):
         self.config = config or {}
         db_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "agent_data.db"))
@@ -880,6 +884,76 @@ class Storage:
             }
         return None
 
+    # Работа с пирам (agent_peers)
+    @staticmethod
+    def parse_hostport(s: str):
+        """
+        Разбирает "IP:port" или "[IPv6]:port" и возвращает (host, port)
+        """
+        s = s.strip()
+        if s.startswith("["):
+            host, _, port = s[1:].partition("]:")
+            try:
+                port = int(port)
+            except:
+                port = None
+            return host, port
+        else:
+            if ":" in s:
+                host, port = s.rsplit(":", 1)
+                try:
+                    port = int(port)
+                except:
+                    port = None
+                return host, port
+            return s, None
+
+    @staticmethod
+    def is_ipv6(host: str):
+        try:
+            socket.inet_pton(socket.AF_INET6, host)
+            return True
+        except OSError:
+            return False
+
+    @staticmethod
+    def is_private(ip: str) -> bool:
+        try:
+            return ipaddress.ip_address(ip).is_private
+        except ValueError:
+            return False
+
+    @classmethod
+    def get_ipv6_scope(cls, host):
+        if host in cls._scope_cache:
+            return cls._scope_cache[host]
+        for iface in netifaces.interfaces():
+            iface_addrs = netifaces.ifaddresses(iface).get(netifaces.AF_INET6, [])
+            for addr_info in iface_addrs:
+                if addr_info.get("addr") == host:
+                    scope_id = socket.if_nametoindex(iface)
+                    cls._scope_cache[host] = scope_id
+                    return scope_id
+        return None
+
+    # Нормализация адресов
+    @classmethod
+    def normalize_address(cls, addr: str) -> str:
+        addr = addr.strip()
+        if not addr:
+            return None
+        if "://" not in addr:
+            addr = f"any://{addr}"
+
+        proto, hostport = addr.split("://", 1)
+        host, port = cls.parse_hostport(hostport)
+
+        # IPv6 без квадратных скобок
+        if cls.is_ipv6(host) and not host.startswith("["):
+            host = f"[{host}]"
+
+        return f"{proto}://{host}:{port}" if port else f"{proto}://{host}"
+
     # Работа с пирам (agent_peers)
     def add_or_update_peer(
         self, peer_id, name, addresses,
@@ -889,18 +963,17 @@ class Storage:
         c = self.conn.cursor()
 
         # нормализация адресов
-        addresses = list({a.strip() for a in (addresses or []) if a and a.strip()})
+        addresses = list({self.normalize_address(a) for a in (addresses or []) if a and a.strip()})
 
-        existing_id = None
         existing_addresses = []
         existing_pubkey = None
         existing_capabilities = {}
 
         if peer_id:
-            c.execute("SELECT id, addresses, pubkey, capabilities FROM agent_peers WHERE id=?", (peer_id,))
+            c.execute("SELECT addresses, pubkey, capabilities FROM agent_peers WHERE id=?", (peer_id,))
             row = c.fetchone()
             if row:
-                existing_id, db_addresses_json, existing_pubkey, db_caps_json = row
+                db_addresses_json, existing_pubkey, db_caps_json = row
                 try:
                     existing_addresses = json.loads(db_addresses_json) or []
                 except:
@@ -910,7 +983,9 @@ class Storage:
                 except:
                     existing_capabilities = {}
 
-        combined_addresses = list({*existing_addresses, *addresses})
+        # объединяем и нормализуем адреса, чтобы IPv6 всегда были с []
+        combined_addresses = list({self.normalize_address(a) for a in (*existing_addresses, *addresses)})
+
         final_pubkey = pubkey or existing_pubkey
         final_capabilities = capabilities or existing_capabilities
 
@@ -937,6 +1012,7 @@ class Storage:
         ))
         self.conn.commit()
 
+    # Получение известных/онлайн пиров
     def get_online_peers(self, limit=50):
         c = self.conn.cursor()
         c.execute("SELECT id, addresses FROM agent_peers WHERE status='online' LIMIT ?", (limit,))
@@ -947,100 +1023,6 @@ class Storage:
         c.execute("SELECT id, addresses FROM agent_peers WHERE id != ? LIMIT ?", (my_id, limit))
         return c.fetchall()
 
-    # Нормализация адресов
-    def normalize_address(self, addr: str) -> str:
-        addr = addr.strip()
-        if not addr:
-            return None
-        if "://" not in addr:
-            return f"any://{addr}"
-        return addr
-
-    # Bootstrap
-    def load_bootstrap(self, bootstrap_file="bootstrap.txt"):
-        """
-        Загружает узлы из bootstrap.txt.
-        Поддерживаются адреса:
-            tcp://host:port
-            udp://host:port
-            any://host:port
-        TCP-узлы проверяются запросом /identity.
-        UDP/any регистрируются без проверки (any учитывается как TCP+UDP).
-        """
-        import requests
-
-        base_path = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-        bootstrap_path = os.path.join(base_path, bootstrap_file)
-
-        if not os.path.exists(bootstrap_path):
-            print(f"[Bootstrap] Файл {bootstrap_file} не найден по пути {bootstrap_path}")
-            return
-
-        for line in open(bootstrap_path, encoding="utf-8"):
-            line = line.strip()
-            if not line or line.startswith("#"):
-                continue
-
-            try:
-                addr = self.normalize_address(line)
-                if addr is None:
-                    continue
-
-                # Разворачиваем any:// на tcp и udp
-                addrs_to_register = []
-                proto, hostport = addr.split("://")
-                if proto == "any":
-                    addrs_to_register.append(f"tcp://{hostport}")
-                    addrs_to_register.append(f"udp://{hostport}")
-                else:
-                    addrs_to_register.append(addr)
-
-                for a in addrs_to_register:
-                    proto2, hostport2 = a.split("://")
-
-                    # TCP → проверяем /identity
-                    if proto2 == "tcp":
-                        try:
-                            url = f"http://{hostport2}/identity"
-                            r = requests.get(url, timeout=3)
-                            if r.status_code == 200:
-                                info = r.json()
-                                peer_id = info.get("id")
-                                name = info.get("name", "unknown")
-                                pubkey = info.get("pubkey")
-                                capabilities = info.get("capabilities", {})
-
-                                self.add_or_update_peer(
-                                    peer_id=peer_id,
-                                    name=name,
-                                    addresses=[a],
-                                    source="bootstrap",
-                                    status="online",
-                                    pubkey=pubkey,
-                                    capabilities=capabilities,
-                                )
-                                print(f"[Bootstrap] Добавлен узел {peer_id} ({a})")
-                            else:
-                                print(f"[Bootstrap] {a} недоступен (HTTP {r.status_code})")
-                        except Exception as e:
-                            print(f"[Bootstrap] Ошибка при подключении к {a}: {e}")
-
-                    # UDP → просто регистрируем
-                    elif proto2 == "udp":
-                        peer_id = str(uuid.uuid4())
-                        self.add_or_update_peer(
-                            peer_id=peer_id,
-                            name="unknown",
-                            addresses=[a],
-                            source="bootstrap",
-                            status="unknown"
-                        )
-                        print(f"[Bootstrap] Добавлен адрес (без проверки): {a}")
-
-            except Exception as e:
-                print(f"[Bootstrap] Ошибка парсинга {line}: {e}")
-
-
     # Утилиты
     def close(self):
         self.conn.close()