Skip to content

Instantly share code, notes, and snippets.

@kamilion
Created November 19, 2018 04:06
Show Gist options
  • Save kamilion/e7f17fa4e0ea2ff3f4105795d9d833e1 to your computer and use it in GitHub Desktop.
Save kamilion/e7f17fa4e0ea2ff3f4105795d9d833e1 to your computer and use it in GitHub Desktop.
Acquire::http::ProxyAutoDetect "/usr/share/squid-deb-proxy-client/apt-avahi-discover";
#!/usr/bin/python
#
# 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)
stdout, stderr = p.communicate()
if not stdout:
DEBUG(
"no timeout set, using default '%s'" % DEFAULT_CONNECT_TIMEOUT_SEC)
return DEFAULT_CONNECT_TIMEOUT_SEC
if not stdout.startswith("TIMEOUT="):
raise ValueError("got unexpected apt-config output: '%s'" % stdout)
varname, sep, value = stdout.strip().partition("=")
timeout = int(value.strip("'"))
DEBUG("using timeout: '%s'" % 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.maxint
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 "<%s> %s: %s" % (
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)
DEBUG("avahi-browse output:")
for line in p.stdout:
DEBUG(" '%s'" % 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: '%s'" % sorted(hosts))
# No host wanted to connect
if (all(h.time_to_connect == sys.maxint 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://[%s]:%s/" % (ip, port)
else:
print "http://%s:%s/" % (ip, port)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment