vilimpoc.org git repositories - dotfiles/blob - xp-fetch.py

   1 #!/usr/bin/env python3
   2 """Download over TLS 1.2 from a machine whose OS cannot.
   3
   4 Windows XP tops out at TLS 1.0, but that limit lives in SChannel, not in the
   5 OS as such: any client carrying its own TLS stack is unaffected. CPython is one
   6 of those - it bundles OpenSSL - so the Python 3.4.4 that setup-windows-xp.bat
   7 installs is a perfectly capable TLS 1.2 client on a box where nothing else is.
   8 That is all this script is: urllib with the protocol floor raised and a CA
   9 bundle supplied from disk.
  10
  11 The CA bundle is not optional. XP's root store predates most roots in use
  12 today, so certificate validation fails against it even when the handshake
  13 itself succeeds. cacert.pem next to this script - Mozilla's set, as published
  14 by the curl project - is what verification actually runs against. A curl or
  15 wget built for XP would need exactly the same thing passed to --cacert.
  16
  17 Usage:
  18     python xp-fetch.py URL [-o FILE]
  19     python xp-fetch.py --tls-check HOST[:PORT]
  20     python xp-fetch.py --tls-check pypi.org --verbose
  21
  22 Written for Python 3.4 (the last release supporting XP) and still correct on
  23 current Pythons, so it can be tested on the host before it is trusted in the
  24 VM.
  25 """
  26
  27 import argparse
  28 import os
  29 import posixpath
  30 import socket
  31 import ssl
  32 import sys
  33 import urllib.error
  34 import urllib.request
  35 from urllib.parse import urlsplit
  36
  37 # Ancient default User-Agents ("Python-urllib/3.4") are turned away by some
  38 # CDNs. Claiming to be a current curl is both honest about being a robot and
  39 # unlikely to be filtered.
  40 USER_AGENT = 'curl/8.0 (windows-xp; xp-fetch.py)'
  41
  42 DEFAULT_CA = 'cacert.pem'
  43 CA_STALE_HINT = ('Refresh it on the host with:\n'
  44                  '    curl -o cacert.pem https://curl.se/ca/cacert.pem')
  45
  46
  47 def find_ca_bundle(explicit):
  48     """Locate the CA bundle: --ca, then next to this script, then the cwd."""
  49     if explicit:
  50         if not os.path.isfile(explicit):
  51             die('CA bundle not found: {0}'.format(explicit))
  52         return explicit
  53     here = os.path.dirname(os.path.abspath(__file__))
  54     for candidate in (os.path.join(here, DEFAULT_CA), DEFAULT_CA):
  55         if os.path.isfile(candidate):
  56             return candidate
  57     die('No {0} found next to {1} or in the current directory.\n{2}'
  58         .format(DEFAULT_CA, os.path.basename(__file__), CA_STALE_HINT))
  59
  60
  61 def build_context(ca_bundle, insecure):
  62     """An SSL context that refuses anything below TLS 1.2.
  63
  64     Nothing here is XP-specific - it is the same context you would want
  65     anywhere. The reason it matters more here is that the fallbacks XP would
  66     otherwise reach for are all dead protocols.
  67     """
  68     if insecure:
  69         context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
  70         context.check_hostname = False
  71         context.verify_mode = ssl.CERT_NONE
  72     else:
  73         context = ssl.create_default_context(cafile=ca_bundle)
  74
  75     # TLSVersion arrived in 3.7; on 3.4 the OP_NO_* flags are the only lever.
  76     minimum = getattr(ssl, 'TLSVersion', None)
  77     if minimum is not None:
  78         context.minimum_version = minimum.TLSv1_2
  79     else:
  80         for flag in ('OP_NO_SSLv2', 'OP_NO_SSLv3', 'OP_NO_TLSv1', 'OP_NO_TLSv1_1'):
  81             context.options |= getattr(ssl, flag, 0)
  82     return context
  83
  84
  85 def tls_check(target, ca_bundle, insecure, verbose):
  86     """Handshake with a host and report what was negotiated.
  87
  88     Worth running first on any new box: it separates "my TLS stack is too old"
  89     from "my certificate store is too old", which produce similar-looking
  90     failures and have completely different fixes.
  91     """
  92     if ':' in target:
  93         host, _, port = target.rpartition(':')
  94         port = int(port)
  95     else:
  96         host, port = target, 443
  97
  98     context = build_context(ca_bundle, insecure)
  99     raw = socket.create_connection((host, port), timeout=30)
 100     try:
 101         with context.wrap_socket(raw, server_hostname=host) as sock:
 102             cipher = sock.cipher() or ('?', '?', 0)
 103             print('host        : {0}:{1}'.format(host, port))
 104             print('protocol    : {0}'.format(cipher[1]))
 105             print('cipher      : {0} ({1} bits)'.format(cipher[0], cipher[2]))
 106             print('verified    : {0}'.format('no (--insecure)' if insecure else 'yes'))
 107             if verbose:
 108                 cert = sock.getpeercert() or {}
 109                 print('subject     : {0}'.format(flatten_name(cert.get('subject'))))
 110                 print('issuer      : {0}'.format(flatten_name(cert.get('issuer'))))
 111                 print('expires     : {0}'.format(cert.get('notAfter', '?')))
 112             print('openssl     : {0}'.format(ssl.OPENSSL_VERSION))
 113     finally:
 114         raw.close()
 115     return 0
 116
 117
 118 def flatten_name(rdns):
 119     if not rdns:
 120         return '?'
 121     parts = []
 122     for rdn in rdns:
 123         for key, value in rdn:
 124             parts.append('{0}={1}'.format(key, value))
 125     return ', '.join(parts)
 126
 127
 128 def output_name(url, explicit):
 129     if explicit:
 130         return explicit
 131     name = posixpath.basename(urlsplit(url).path)
 132     return name or 'index.html'
 133
 134
 135 def fetch(url, dest, ca_bundle, insecure, verbose):
 136     context = build_context(ca_bundle, insecure)
 137     opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context))
 138     request = urllib.request.Request(url, headers={'User-Agent': USER_AGENT})
 139
 140     response = opener.open(request, timeout=60)
 141     try:
 142         if verbose:
 143             print('{0} {1}'.format(response.status if hasattr(response, 'status')
 144                                    else response.getcode(), url), file=sys.stderr)
 145         total = response.headers.get('Content-Length')
 146         total = int(total) if total and total.isdigit() else None
 147         written = 0
 148         with open(dest, 'wb') as out:
 149             while True:
 150                 chunk = response.read(64 * 1024)
 151                 if not chunk:
 152                     break
 153                 out.write(chunk)
 154                 written += len(chunk)
 155                 report(written, total)
 156     finally:
 157         response.close()
 158     sys.stderr.write('\n')
 159     print('{0}  ({1} bytes)'.format(dest, written))
 160
 161     if total is not None and written != total:
 162         die('Truncated: expected {0} bytes, got {1}.'.format(total, written), code=3)
 163     return 0
 164
 165
 166 def report(written, total):
 167     """One rewritten line of progress. XP consoles are slow; keep it cheap.
 168
 169     Only when stderr is a console: redirected to a file or a pipe, \r does not
 170     rewrite anything and every tick lands as another line of noise in the log.
 171     """
 172     if not sys.stderr.isatty():
 173         return
 174     if total:
 175         sys.stderr.write('\r  {0:>10} / {1} bytes ({2:3.0f}%)'
 176                          .format(written, total, 100.0 * written / total))
 177     else:
 178         sys.stderr.write('\r  {0:>10} bytes'.format(written))
 179     sys.stderr.flush()
 180
 181
 182 def die(message, code=2):
 183     sys.stderr.write('xp-fetch: {0}\n'.format(message))
 184     raise SystemExit(code)
 185
 186
 187 def main(argv=None):
 188     parser = argparse.ArgumentParser(
 189         description='Fetch over TLS 1.2 on a host whose OS TLS stack cannot.')
 190     parser.add_argument('url', nargs='?', help='URL to download')
 191     parser.add_argument('-o', '--output', help='output file (default: name from the URL)')
 192     parser.add_argument('--ca', help='CA bundle (default: cacert.pem beside this script)')
 193     parser.add_argument('--tls-check', metavar='HOST[:PORT]',
 194                         help='handshake only, and report protocol and cipher')
 195     parser.add_argument('--insecure', action='store_true',
 196                         help='skip certificate verification (diagnostics only)')
 197     parser.add_argument('-v', '--verbose', action='store_true')
 198     args = parser.parse_args(argv)
 199
 200     if not args.url and not args.tls_check:
 201         parser.error('give a URL, or --tls-check HOST')
 202
 203     ca_bundle = None if args.insecure else find_ca_bundle(args.ca)
 204     if args.verbose and ca_bundle:
 205         print('ca bundle   : {0}'.format(ca_bundle), file=sys.stderr)
 206
 207     try:
 208         if args.tls_check:
 209             return tls_check(args.tls_check, ca_bundle, args.insecure, args.verbose)
 210         return fetch(args.url, output_name(args.url, args.output),
 211                      ca_bundle, args.insecure, args.verbose)
 212     except ssl.SSLError as exc:
 213         die('TLS failed: {0}\n'
 214             'If this says certificate verify failed, the CA bundle is stale or wrong.\n{1}'
 215             .format(exc, CA_STALE_HINT), code=4)
 216     except (urllib.error.URLError, socket.error) as exc:
 217         die('{0}'.format(exc), code=5)
 218
 219
 220 if __name__ == '__main__':
 221     sys.exit(main())