-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdns_service.py
More file actions
89 lines (76 loc) · 2.84 KB
/
Copy pathdns_service.py
File metadata and controls
89 lines (76 loc) · 2.84 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import socket
import logging
from zeroconf import Zeroconf, ServiceInfo
class DNSService:
def __init__(self, port, aliases=None):
self.port = port
self.aliases = aliases or []
self.logger = logging.getLogger("DNSService")
# Get local IP first to bind specifically to it
self.ip_address = self.get_local_ip()
print(f"[DNSService] Binding mDNS to interface: {self.ip_address}")
# Explicitly bind to the specific interface
try:
self.zeroconf = Zeroconf(interfaces=[self.ip_address])
except Exception:
self.zeroconf = Zeroconf() # Fallback to default
self.infos = []
def register(self):
"""Register all aliases as mDNS services."""
ip_address = self.ip_address # Use the IP we bound to
try:
# We need the IP in bytes for zeroconf
ip_bytes = socket.inet_aton(ip_address)
except Exception as e:
print(f"Failed to convert IP {ip_address}: {e}")
return
for alias in self.aliases:
# Service type is _http._tcp.local.
# The instance name needs to be unique on the network, usually.
# We'll use the alias as the instance name.
name = f"{alias}._http._tcp.local."
# Host must be fully qualified name, e.g. alias.local.
server_name = f"{alias}.local."
info = ServiceInfo(
"_http._tcp.local.",
name,
addresses=[ip_bytes],
port=self.port,
server=server_name,
properties={'desc': 'SajiloCloud File Server'}
)
try:
self.zeroconf.register_service(info)
self.infos.append(info)
print(f"[DNSService] Registered http://{alias}.local:{self.port}")
except Exception as e:
print(f"[DNSService] Failed to register {alias}: {e}")
def unregister(self):
"""Unregister all services."""
for info in self.infos:
try:
self.zeroconf.unregister_service(info)
except Exception:
pass
self.zeroconf.close()
print("[DNSService] Unregistered all services")
def get_local_ip(self):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
except Exception:
ip = "127.0.0.1"
finally:
s.close()
return ip
if __name__ == "__main__":
# Test
service = DNSService(4142, ["test-server", "sajilo"])
service.register()
import time
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
service.unregister()