Upload a TLS key and cert to a FRITZ!Box, in pretty Python

fritz_cert_upload.py

#!/usr/bin/env python
# vim: fileencoding=utf-8
"""
Upload a TLS key and cert to a FRITZ!Box, in pretty Python
Copyright (C) 2018 Olivier Mehani <[email protected]>

This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.
"""

from hashlib import md5
import requests
import re
import logging

class FritzBoxAuthService:
    '''A service to authenticate and retrieve a session ID from a FRITZ!Box

    Based on hsmade/get_sid.py [0].

    [0] https://gist.github.com/hsmade/b9e8aed176be28abe7a2e5bdbec52a8d'''

    INVALID_SID = '0000000000000000'

    _logger = None

    def __init__(self):
        self._logger = logging.getLogger(self.__class__.__name__)

    def set_logger(self, logger):
        self._logger = logger

    def password_auth(self, host, username, password):
        '''Authenticate using a password to get a token, returns a SID'''
        challenge = self.get_challenge(host)
        token = self.compute_token(challenge, password)
        return self.token_auth(host, username, token)

    def get_challenge(self, host):
      '''Get the challenge, do very ugly xml parsing'''
      response = requests.get('http://{host}/login_sid.lua'.format(host=host))
      self._logger.debug('response: %s' % (response.text))
      challenge = response.content.split('Challenge>')[1].split('<')[0]
      self._logger.debug('challenge: %s' % (challenge))
      return challenge

    def _u16le_nobom(self, str):
        return str.decode('utf8').encode('utf16')[2:]

    def compute_token(self, challenge, password):
      '''Calculate the response token

        See [0], example from p. 3

        [0] https://avm.de/fileadmin/user_upload/Global/Service/Schnittstellen/AVM_Technical_Note_-_Session_ID.pdf

        >>> auth_service = FritzBoxAuthService()
        >>> auth_service.compute_token('1234567z', 'äbc')
        '1234567z-9e224a41eeefa284df7bb0f26c2913e2'

      '''
      hash = md5(
          '{challenge}{dash}{password}'.format(
              challenge=self._u16le_nobom(challenge),
              dash=self._u16le_nobom('-'),
              password=self._u16le_nobom(password),
          )
      ).hexdigest()
      token = '{challenge}-{hash}'.format(
          challenge=challenge,
          hash=hash,
      )
      self._logger.debug('token: %s' % (token,))
      return token

    def token_auth(self, host, username, token):
        '''Authenticate using a token, returns a SID'''
        payload = {
            'username': username,
            'response': token,
        }
        response = requests.post('http://{host}/login_sid.lua'.format(host=host),
                                 data=payload)
        self._logger.debug('response: %s' % (response.text))
        sid = response.content.split('<SID>')[1].split('<')[0]
        self._logger.debug('SID: %s' % (sid,))
        if sid == self.INVALID_SID:
            raise FritzBoxLoginException('Failed to get valid SID from host "%s"' %
                                         (host,)
                                         )
        return sid


class FritzBoxLoginException(Exception):
    pass


class FritzBoxCertService:
    '''Service allowing to upload a TLS key and certificate to a FRITZ!Box

    With inspiration from wikrie/fritzbox-cert-update.sh [0].

    [0] https://gist.github.com/wikrie/f1d5747a714e0a34d0582981f7cb4cfb'''

    _logger = None

    def __init__(self):
        self._logger = logging.getLogger(self.__class__.__name__)

    def set_logger(self, logger):
        self._logger = logger

    def upload_key_cert(self, host, sid, key, cert, key_password=None):
        '''Upload the key and certificate using a valid SID'''
        params = {
            'sid': sid,
        }
        if key_password is not None:
            # XXX: If it is unconditionally included, even if empty,
            # this seems to throw the upload off. Could also be an ordering issue,
            # as request puts this parameters first, before the sid
            params['BoxCertPassword'] = key_password
        certfile = {
            'BoxCertImportFile': (
                'BoxCert.pem',
                '{key}{cert}'.format(
                    key=key,
                    cert=cert
                ),
                'application/x-x509-ca-cert',
            ),
        }
        response = requests.post(
            'http://{host}/cgi-bin/firmwarecfg'.format(host=host),
            data=params,
            files=certfile
        )
        if response.status_code != 200:
            raise FritzBoxCertUploadException('Failed to upload certificate to host "%s"' %
                                              (host,)
                                              )
        for line in response.text.split('\n'):
            if re.search('SSL', line):
                return line
            elif re.search('ErrorMsg', line):
                raise FritzBoxCertUploadException(
                    line.split('"ErrorMsg">')[1].split('</')[0]
                )
        raise FritzBoxCertUploadException('Uploaded certificate, but the FRITZ!Box did not acknowledge the attempt')


class FritzBoxCertUploadException(Exception):
    pass


if __name__ == '__main__':
    import argparse
    from getpass import getpass
    import sys

    def parse_arguments():
        LE_PATH='/etc/letsencrypt/live/{domain}'
        parser = argparse.ArgumentParser(description='Upload Let\'s Encrypt certificate to FRITZ!Box',
                                         epilog='example: %s -u USER -P -d fritzbox.example.net' % sys.argv[0])
        parser.add_argument('-D', '--debug-level', default='warning')
        parser.add_argument('-H', '--host', default='fritz.box')
        parser.add_argument('-u', '--username', default=None)
        parser.add_argument('-p', '--password', default=None)
        parser.add_argument('-P', '--ask-password', default=False, nargs='?', const=True)
        parser.add_argument('-d', '--domain',
                            help='Look for Let\'s Encrypt keys in ' + LE_PATH.format(domain='DOMAIN'))
        parser.add_argument('-k', '--key-file')
        parser.add_argument('-f', '--key-passphrase', default=None)
        parser.add_argument('-F', '--ask-key-passphrase', default=False, nargs='?', const=True)
        parser.add_argument('-c', '--cert-file')
        args = vars(parser.parse_args())

        if args['ask_password']:
            args['password'] = getpass('Password for {username}@{host}: '.format(
                username=args['username'],
                host=args['host']
            ))

        if args['domain'] is not None:
            certpath = LE_PATH.format(domain=args['domain'])
            args['key_file'] = '{certpath}/privkey.pem'.format(certpath=certpath)
            args['cert_file'] = '{certpath}/fullchain.pem'.format(certpath=certpath)

        if args['key_file'] is None or args['cert_file'] is None:
            logging.error('Neither Key and certificate, nor domain were specified')
            sys.exit(1)

        if args['ask_key_passphrase']:
            args['key_passphrase'] = getpass('Passphrase for {key}: '.format(
                key=args['key_file']
            ))

        if args['username'] is None or args['password'] is None:
            logging.error('Username or password were specified')
            sys.exit(1)

        return args


    args = parse_arguments()

    logging.basicConfig(level=args['debug_level'].upper())
    logging.info('FRITZ!Box Certificate upload script')

    auth_service = FritzBoxAuthService()
    auth_service.set_logger(logging.getLogger('authService'))
    sid = auth_service.password_auth(
        args['host'],
        args['username'],
        args['password']
    )
    with open(args['key_file']) as key:
        with open(args['cert_file']) as cert:
            cert_service = FritzBoxCertService()
            cert_service.set_logger(logging.getLogger('certService'))
            result = cert_service.upload_key_cert(args['host'],
                                                sid,
                                                key.read(),
                                                cert.read(),
                                                args['key_passphrase'])
            print(result)