File: apt-avahi-discover

package info (click to toggle)
squid-deb-proxy 0.8.15%2Bnmu1
  • links: PTS
  • area: main
  • in suites: bookworm, sid
  • size: 252 kB
  • sloc: sh: 315; ansic: 145; python: 100; makefile: 32
file content (132 lines) | stat: -rwxr-xr-x 4,189 bytes parent folder | download | duplicates (2)
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#!/usr/bin/python3
#
# use avahi to find a _apt_proxy._tcp provider and return
# a http proxy string suitable for apt

import asyncore
import functools
import os
import socket
import sys
import time
from subprocess	import Popen, PIPE, call

DEFAULT_CONNECT_TIMEOUT_SEC = 2

def DEBUG(msg):
    if "--debug" in sys.argv:
        sys.stderr.write(msg + "\n")


def get_avahi_discover_timeout():
    APT_AVAHI_TIMEOUT_VAR = "APT::Avahi-Discover::Timeout"
    p = Popen(
        ["/usr/bin/apt-config", "shell", "TIMEOUT", APT_AVAHI_TIMEOUT_VAR], 
        stdout=PIPE, universal_newlines=True)
    stdout, stderr = p.communicate()
    if not stdout:
        DEBUG(
            "no timeout set, using default '{}'".format(DEFAULT_CONNECT_TIMEOUT_SEC))
        return DEFAULT_CONNECT_TIMEOUT_SEC
    if not stdout.startswith("TIMEOUT="):
        raise ValueError("got unexpected apt-config output: '{}'".format(stdout))
    varname, sep, value = stdout.strip().partition("=")
    timeout = int(value.strip("'"))
    DEBUG("using timeout: '{}'".format(timeout))
    return timeout

@functools.total_ordering
class AptAvahiClient(asyncore.dispatcher):
    def __init__(self, addr):
        asyncore.dispatcher.__init__(self)
        if is_ipv6(addr[0]):
            self.create_socket(socket.AF_INET6, socket.SOCK_STREAM)
            self.connect( (addr[0], addr[1], 0, 0) )
        else:
            self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
            self.connect(addr)
        self._time_init = time.time()
        self.time_to_connect = sys.maxsize
        self.address = addr
    def handle_connect(self):
        self.time_to_connect = time.time() - self._time_init
        self.close()
    def __eq__(self, other):
        return self.time_to_connect == other.time_to_connect
    def __lt__(self, other):
        return self.time_to_connect < other.time_to_connect
    def __repr__(self):
        return "<{}> {}: {}".format(
            self.__class__.__name__, self.addr, self.time_to_connect)


def is_ipv6(a):
    return ':' in a

def is_linklocal(addr):
    # Link-local should start with fe80 and six null bytes
    return addr.startswith("fe80::")

def get_proxy_host_port_from_avahi():
    service = '_apt_proxy._tcp'

    # Obtain all of the services addresses from avahi, pulling the IPv6
    # addresses to the top.
    addr4 = []
    addr6 = []
    p = Popen(['avahi-browse', '-kprtf', service], stdout=PIPE, universal_newlines=True)
    DEBUG("avahi-browse output:")
    for line in p.stdout:
        DEBUG(" '{}'".format(line))
        if line.startswith('='):
            tokens = line.split(';')
            addr = tokens[7]
            port = int(tokens[8])
            if is_ipv6(addr):
                # We need to skip ipv6 link-local addresses since 
                # APT can't use them
                if not is_linklocal(addr):
                    addr6.append((addr, port))
            else:
                addr4.append((addr, port))

    # Run through the offered addresses and see if we we have a bound local
    # address for it.
    addrs = []
    for (ip, port) in addr6 + addr4:
        try:
            res = socket.getaddrinfo(ip, port, 0, 0, 0, socket.AI_ADDRCONFIG)
            if res:
                addrs.append((ip, port))
        except socket.gaierror:
            pass
    if not addrs:
        return None
    
    # sort by answering speed
    hosts = []
    for addr in addrs:
        hosts.append(AptAvahiClient(addr))
    # 2s timeout, arbitray
    timeout = get_avahi_discover_timeout()
    asyncore.loop(timeout=timeout)
    DEBUG("sorted hosts: '{}'".format(sorted(hosts)))

    # No host wanted to connect
    if (all(h.time_to_connect == sys.maxsize for h in hosts)):
        return None

    fastest_host = sorted(hosts)[0]
    fastest_address = fastest_host.address
    return fastest_address


if __name__ == "__main__":
    # Dump the approved address out in an appropriate format.
    address = get_proxy_host_port_from_avahi()
    if address:
        (ip, port) = address
        if is_ipv6(ip):
            print("http://[{}]:{}/".format(ip, port))
        else:
            print("http://{}:{}/".format(ip, port))