dotfiles: Windows XP VM provisioning

[dotfiles] / xp-fetch.py

#!/usr/bin/env python3
"""Download over TLS 1.2 from a machine whose OS cannot.

Windows XP tops out at TLS 1.0, but that limit lives in SChannel, not in the
OS as such: any client carrying its own TLS stack is unaffected. CPython is one
of those - it bundles OpenSSL - so the Python 3.4.4 that setup-windows-xp.bat
installs is a perfectly capable TLS 1.2 client on a box where nothing else is.
That is all this script is: urllib with the protocol floor raised and a CA
bundle supplied from disk.

The CA bundle is not optional. XP's root store predates most roots in use
today, so certificate validation fails against it even when the handshake
itself succeeds. cacert.pem next to this script - Mozilla's set, as published
by the curl project - is what verification actually runs against. A curl or
wget built for XP would need exactly the same thing passed to --cacert.

Usage:
    python xp-fetch.py URL [-o FILE]
    python xp-fetch.py --tls-check HOST[:PORT]
    python xp-fetch.py --tls-check pypi.org --verbose

Written for Python 3.4 (the last release supporting XP) and still correct on
current Pythons, so it can be tested on the host before it is trusted in the
VM.
"""

import argparse
import os
import posixpath
import socket
import ssl
import sys
import urllib.error
import urllib.request
from urllib.parse import urlsplit

# Ancient default User-Agents ("Python-urllib/3.4") are turned away by some
# CDNs. Claiming to be a current curl is both honest about being a robot and
# unlikely to be filtered.
USER_AGENT = 'curl/8.0 (windows-xp; xp-fetch.py)'

DEFAULT_CA = 'cacert.pem'
CA_STALE_HINT = ('Refresh it on the host with:\n'
                 '    curl -o cacert.pem https://curl.se/ca/cacert.pem')


def find_ca_bundle(explicit):
    """Locate the CA bundle: --ca, then next to this script, then the cwd."""
    if explicit:
        if not os.path.isfile(explicit):
            die('CA bundle not found: {0}'.format(explicit))
        return explicit
    here = os.path.dirname(os.path.abspath(__file__))
    for candidate in (os.path.join(here, DEFAULT_CA), DEFAULT_CA):
        if os.path.isfile(candidate):
            return candidate
    die('No {0} found next to {1} or in the current directory.\n{2}'
        .format(DEFAULT_CA, os.path.basename(__file__), CA_STALE_HINT))


def build_context(ca_bundle, insecure):
    """An SSL context that refuses anything below TLS 1.2.

    Nothing here is XP-specific - it is the same context you would want
    anywhere. The reason it matters more here is that the fallbacks XP would
    otherwise reach for are all dead protocols.
    """
    if insecure:
        context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
    else:
        context = ssl.create_default_context(cafile=ca_bundle)

    # TLSVersion arrived in 3.7; on 3.4 the OP_NO_* flags are the only lever.
    minimum = getattr(ssl, 'TLSVersion', None)
    if minimum is not None:
        context.minimum_version = minimum.TLSv1_2
    else:
        for flag in ('OP_NO_SSLv2', 'OP_NO_SSLv3', 'OP_NO_TLSv1', 'OP_NO_TLSv1_1'):
            context.options |= getattr(ssl, flag, 0)
    return context


def tls_check(target, ca_bundle, insecure, verbose):
    """Handshake with a host and report what was negotiated.

    Worth running first on any new box: it separates "my TLS stack is too old"
    from "my certificate store is too old", which produce similar-looking
    failures and have completely different fixes.
    """
    if ':' in target:
        host, _, port = target.rpartition(':')
        port = int(port)
    else:
        host, port = target, 443

    context = build_context(ca_bundle, insecure)
    raw = socket.create_connection((host, port), timeout=30)
    try:
        with context.wrap_socket(raw, server_hostname=host) as sock:
            cipher = sock.cipher() or ('?', '?', 0)
            print('host        : {0}:{1}'.format(host, port))
            print('protocol    : {0}'.format(cipher[1]))
            print('cipher      : {0} ({1} bits)'.format(cipher[0], cipher[2]))
            print('verified    : {0}'.format('no (--insecure)' if insecure else 'yes'))
            if verbose:
                cert = sock.getpeercert() or {}
                print('subject     : {0}'.format(flatten_name(cert.get('subject'))))
                print('issuer      : {0}'.format(flatten_name(cert.get('issuer'))))
                print('expires     : {0}'.format(cert.get('notAfter', '?')))
            print('openssl     : {0}'.format(ssl.OPENSSL_VERSION))
    finally:
        raw.close()
    return 0


def flatten_name(rdns):
    if not rdns:
        return '?'
    parts = []
    for rdn in rdns:
        for key, value in rdn:
            parts.append('{0}={1}'.format(key, value))
    return ', '.join(parts)


def output_name(url, explicit):
    if explicit:
        return explicit
    name = posixpath.basename(urlsplit(url).path)
    return name or 'index.html'


def fetch(url, dest, ca_bundle, insecure, verbose):
    context = build_context(ca_bundle, insecure)
    opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context))
    request = urllib.request.Request(url, headers={'User-Agent': USER_AGENT})

    response = opener.open(request, timeout=60)
    try:
        if verbose:
            print('{0} {1}'.format(response.status if hasattr(response, 'status')
                                   else response.getcode(), url), file=sys.stderr)
        total = response.headers.get('Content-Length')
        total = int(total) if total and total.isdigit() else None
        written = 0
        with open(dest, 'wb') as out:
            while True:
                chunk = response.read(64 * 1024)
                if not chunk:
                    break
                out.write(chunk)
                written += len(chunk)
                report(written, total)
    finally:
        response.close()
    sys.stderr.write('\n')
    print('{0}  ({1} bytes)'.format(dest, written))

    if total is not None and written != total:
        die('Truncated: expected {0} bytes, got {1}.'.format(total, written), code=3)
    return 0


def report(written, total):
    """One rewritten line of progress. XP consoles are slow; keep it cheap.

    Only when stderr is a console: redirected to a file or a pipe, \r does not
    rewrite anything and every tick lands as another line of noise in the log.
    """
    if not sys.stderr.isatty():
        return
    if total:
        sys.stderr.write('\r  {0:>10} / {1} bytes ({2:3.0f}%)'
                         .format(written, total, 100.0 * written / total))
    else:
        sys.stderr.write('\r  {0:>10} bytes'.format(written))
    sys.stderr.flush()


def die(message, code=2):
    sys.stderr.write('xp-fetch: {0}\n'.format(message))
    raise SystemExit(code)


def main(argv=None):
    parser = argparse.ArgumentParser(
        description='Fetch over TLS 1.2 on a host whose OS TLS stack cannot.')
    parser.add_argument('url', nargs='?', help='URL to download')
    parser.add_argument('-o', '--output', help='output file (default: name from the URL)')
    parser.add_argument('--ca', help='CA bundle (default: cacert.pem beside this script)')
    parser.add_argument('--tls-check', metavar='HOST[:PORT]',
                        help='handshake only, and report protocol and cipher')
    parser.add_argument('--insecure', action='store_true',
                        help='skip certificate verification (diagnostics only)')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args(argv)

    if not args.url and not args.tls_check:
        parser.error('give a URL, or --tls-check HOST')

    ca_bundle = None if args.insecure else find_ca_bundle(args.ca)
    if args.verbose and ca_bundle:
        print('ca bundle   : {0}'.format(ca_bundle), file=sys.stderr)

    try:
        if args.tls_check:
            return tls_check(args.tls_check, ca_bundle, args.insecure, args.verbose)
        return fetch(args.url, output_name(args.url, args.output),
                     ca_bundle, args.insecure, args.verbose)
    except ssl.SSLError as exc:
        die('TLS failed: {0}\n'
            'If this says certificate verify failed, the CA bundle is stale or wrong.\n{1}'
            .format(exc, CA_STALE_HINT), code=4)
    except (urllib.error.URLError, socket.error) as exc:
        die('{0}'.format(exc), code=5)


if __name__ == '__main__':
    sys.exit(main())