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

fritz_cert_upload.py

#!/usr/bin/env python3
# vim: fileencoding=utf-8
"""
Upload a TLS key and cert to a FRITZ!Box, in pretty Python
Copyright (C) 2018--2021 Olivier Mehani <shtrom@ssji.net>

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'

    CHALLENGE_RE = b'.*<Challenge>([^<]+)</Challenge>.*'
    SID_RE = b'.*<SID>([^<]+)</SID>.*'

    _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)).content
        challenge_match = re.match(self.CHALLENGE_RE, response)
        if not challenge_match:
            self._logger.debug(challenge_match)
            raise FritzBoxLoginException(
                'Failed to get challenge from host: {response}')
        challenge = challenge_match[1].decode('utf-8')
        self._logger.debug(f'Found challenge: {challenge}')
        return challenge

    def _u16le_nobom(self, str):
        return str.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_toen('1234567z', 'äbc')
          '1234567z-9e224a41eeefa284df7bb0f26c2913e2'
          >>> auth_service.compute_token('', '0[Rzschc<YD@[DV0jsR>')
          ''

        '''
        hash = md5(
            self._u16le_nobom(
                f'{challenge}-{password}'
            )

        ).hexdigest()
        token = f'{challenge}-{hash}'
        self._logger.debug(f'Auth token: {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).content
        sid_match = re.match(self.SID_RE, response, re.MULTILINE)
        if not sid_match:
            raise FritzBoxLoginException(
                f'Failed to get SID from host: {response}')
        sid = sid_match[1].decode('utf-8')
        self._logger.debug(f'Found SID: {sid}')
        if sid == self.INVALID_SID:
            raise FritzBoxLoginException(
                f'Failed to get authenticate to host `{host}` (received invalid SID after authentication)')
        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'''

    ERROR_RE = '.*<ErrorMsg>([^<]+)</ErrorMsg>.*'

    _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(
                f'Failed to upload certificate to host `{host}`')
        response_text = response.text
        for line in response_text.split('\n'):
            if re.search('SSL', line):
                return line
            elif error_message := re.match(self.ERROR_RE, line, re.MULTILINE):
                raise FritzBoxCertUploadException(
                    error_message
                )
        raise FritzBoxCertUploadException(
            f'Uploaded certificate, but the FRITZ!Box did not acknowledge the attempt: {response_text}')


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=f'example: {sys.argv[0]} -u USER -d fritzbox.example.net')
        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('-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 not args['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 not specified')
            sys.exit(1)

        return args

    def upload_cert(logger, host, username, password, key_file, cert_file, key_passphrase):
        auth_service = FritzBoxAuthService()
        auth_service.set_logger(logger.getChild('AuthService'))
        sid = auth_service.password_auth(host, username, password)

        with open(key_file) as key:
            with open(cert_file) as cert:
                cert_service = FritzBoxCertService()
                cert_service.set_logger(logger.getChild('CertService'))
                result = cert_service.upload_key_cert(host,
                                                      sid,
                                                      key.read(),
                                                      cert.read(),
                                                      key_passphrase)
                logger.info(result)

    args = parse_arguments()

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

    try:
        upload_cert(
            logger,
            args['host'],
            args['username'],
            args['password'],
            args['key_file'],
            args['cert_file'],
            args['key_passphrase']
        )
    except FritzBoxLoginException as e:
        logger.error(e)