FreedomBox/actions/tor

#!/usr/bin/python3
# -*- mode: python -*-
#
# This file is part of Plinth.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

"""
Configuration helper for the Tor service
"""

import argparse
import codecs
import os
import re
import socket

from plinth import action_utils

TOR_CONFIG = '/etc/tor/torrc'
TOR_STATE_FILE = '/var/lib/tor/state'
TOR_AUTH_COOKIE = '/var/run/tor/control.authcookie'


def parse_arguments():
    """Return parsed command line arguments as dictionary"""
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')

    subparsers.add_parser('enable', help='Enable and start Tor service')
    subparsers.add_parser('disable', help='Disable and stop Tor service')
    subparsers.add_parser('get-hs', help='Get hidden service')
    subparsers.add_parser('enable-hs', help='Enable hidden service')
    subparsers.add_parser('disable-hs', help='Disable hidden service')
    subparsers.add_parser('get-ports', help='Get list of Tor ports')

    return parser.parse_args()


def subcommand_enable(_):
    """Enable and start the service."""
    action_utils.service_enable('tor')


def subcommand_disable(_):
    """Disable and stop the service."""
    action_utils.service_disable('tor')


def subcommand_get_hs(_):
    """Print currently configured Tor hidden service information"""
    print(get_hidden_service())


def subcommand_enable_hs(_):
    """Enable Tor hidden service"""
    if get_hidden_service():
        return

    with open(TOR_CONFIG, 'r') as conffile:
        lines = conffile.readlines()

    lines.append('# Hidden Service configured by Plinth\n')
    lines.append('HiddenServiceDir /var/lib/tor/hidden_service/\n')
    lines.append('HiddenServicePort 80 127.0.0.1:80\n')
    lines.append('HiddenServicePort 443 127.0.0.1:443\n')
    lines.append('# end of Plinth Hidden Service config\n')

    with open(TOR_CONFIG, 'w') as conffile:
        conffile.writelines(lines)

    action_utils.service_restart('tor')


def subcommand_disable_hs(_):
    """Disable Tor hidden service"""
    if not get_hidden_service():
        return

    with open(TOR_CONFIG, 'r') as conffile:
        lines = conffile.readlines()

        filtered_lines = []
        removing = False
        for line in lines:
            if removing:
                if line.startswith('# end of Plinth Hidden Service config'):
                    # last line of Plinth hidden service block
                    # stop removing after this line
                    removing = False
                elif not line.startswith('HiddenService'):
                    # end of Plinth hidden service block
                    # stop removing lines
                    removing = False
                    filtered_lines.append(line)
            else:
                if line.startswith('# Hidden Service configured by Plinth'):
                    # start of Plinth hidden service block
                    # remove following HiddenService lines
                    removing = True
                else:
                    filtered_lines.append(line)

    with open(TOR_CONFIG, 'w') as conffile:
        conffile.writelines(filtered_lines)

    action_utils.service_restart('tor')


def get_hidden_service():
    """Return a string with configured Tor hidden service information"""
    hs_dir = None
    hs_ports = []

    with open(TOR_CONFIG, 'r') as conf_file:
        for line in conf_file:
            if line.startswith('HiddenServiceDir'):
                hs_dir = line.split()[1]
            elif line.startswith('HiddenServicePort'):
                hs_ports.append(line.split()[1])

    if not hs_dir:
        return ''

    try:
        with open(os.path.join(hs_dir, 'hostname'), 'r') as conf_file:
            hs_hostname = conf_file.read().strip()
    except Exception:
        return 'error'

    return hs_hostname + ' ' + ','.join(hs_ports)


def subcommand_get_ports(_):
    """Return a list of running Tor ports."""
    try:
        print('orport', _get_orport())
    except Exception:
        pass

    with open(TOR_STATE_FILE, 'r') as state_file:
        for line in state_file:
            matches = re.match(r'^\s*TransportProxy\s+(\S*)\s+\S+:(\d+)\s*$',
                               line)
            if matches:
                print('{0} {1}'.format(matches.group(1), matches.group(2)))


def _get_orport():
    """Return the ORPort by querying running instance."""
    cookie = open(TOR_AUTH_COOKIE, 'rb').read()
    cookie = codecs.encode(cookie, 'hex').decode()

    commands = '''AUTHENTICATE {cookie}
GETINFO net/listeners/or
QUIT
'''.format(cookie=cookie)

    tor_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    tor_socket.connect(('localhost', 9051))
    tor_socket.send(commands.encode())
    response = tor_socket.recv(1024)
    tor_socket.close()

    line = response.split(b'\r\n')[1].decode()
    matches = re.match(r'.*="[^:]+:(\d+)"', line)
    return matches.group(1)


def main():
    """Parse arguments and perform all duties"""
    arguments = parse_arguments()

    subcommand = arguments.subcommand.replace('-', '_')
    subcommand_method = globals()['subcommand_' + subcommand]
    subcommand_method(arguments)


if __name__ == '__main__':
    main()