outline-cli-wrapper.py

import shutil
import base64
import sys
import os
import http.client
import urllib.parse
import json
import logging

# Url for retrieving shadowsocks configuration, allowed schemas: none, ssconf, http, https (default)
DEFAULT_URL = 'ssconf://default.url'
# Full path to outline-cli executable
OUTLINE_CLI_PATH = '/usr/bin/outline-cli'

logging.basicConfig(
    level=logging.DEBUG,
    format="[%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()],
    force=True
)
logger = logging.getLogger(__name__)


def get_config_from_url(parsed_url):
    try:
        # Determine the connection type (HTTP or HTTPS)
        if parsed_url.scheme == 'https':
            conn = http.client.HTTPSConnection(parsed_url.netloc)
        else:
            conn = http.client.HTTPConnection(parsed_url.netloc)

        path = parsed_url.path or "/"

        conn.request("GET", path)

        # Get the response
        response = conn.getresponse()
        logger.debug(f"Status Code: {response.status}")

        # Read and decode the response
        response_data = response.read().decode()

        # Attempt to parse the response as JSON
        try:
            json_data = json.loads(response_data)
            logger.debug("JSON Response Content:")
            logger.debug(json.dumps(json_data, indent=4))  # Pretty print the JSON data
            return json_data
        except json.JSONDecodeError:
            logger.debug("Failed to parse response as JSON. Raw response content:")
            logger.debug(response_data)
        finally:
            conn.close()

    except Exception as e:
        logger.error(f"An error occurred: {e}")


def convert_to_config_string(json_map):
    """https://shadowsocks.org/doc/sip002.html
    SS-URI = "ss://" userinfo "@" hostname ":" port [ "/" ] [ "?" plugin ] [ "#" tag ]
    userinfo = websafe-base64-encode-utf8(method  ":" password)
           method ":" password
    """

    method = json_map.get('method')
    password = json_map.get('password')
    server = json_map.get('server')
    server_port = json_map.get('server_port')
    user_info = base64.urlsafe_b64encode(f"{method}:{password}".encode("utf-8")).decode()

    return f"{user_info}@{server}:{server_port}"


def normalize_url(url_to_normalize):
    if url_to_normalize.startswith('ssconf://'):
        normalized_url = url_to_normalize.replace('ssconf://', 'https://')
    elif not url_to_normalize.startswith(('http://', 'https://')):
        normalized_url = 'https://' + url_to_normalize
    else:
        normalized_url = url_to_normalize
    return normalized_url


def parse_url(url_to_parse):
    parsed_url = urllib.parse.urlparse(url_to_parse)
    logger.debug(f"{parsed_url}")
    return parsed_url


def get_ss_url(ssconf_url):
    logger.debug(f"url: [{ssconf_url}]")
    url_object = parse_url(normalize_url(ssconf_url))
    outline_cli_config_str = convert_to_config_string(get_config_from_url(url_object))
    logger.debug(outline_cli_config_str)
    return f"ss://{outline_cli_config_str}"


def launch_application(app_path, app_name, args=[]):
    logger.info(f"{app_path} {args}")
    if app_path and os.access(app_path, os.X_OK):
        os.execv(app_path, [app_name] + args)


def launch_outline_cli(ss_url):
    app_name = 'outline-cli'
    args = ['-transport', ss_url]

    app_path = shutil.which(app_name)
    if not app_path:
        app_path = OUTLINE_CLI_PATH
    launch_application(app_path, app_name, args)


if __name__ == "__main__":
    """
    This script retrieves properties from ssconf:// url and launches specified instance of outline_cli
    Might have to be run with superuser/admin privileges
    """

    if len(sys.argv) != 2:
        url = os.environ.get('OUTLINE_CLI_SSCONF_URI', DEFAULT_URL)
    else:
        url = sys.argv[1]
    logger.info(f"Using following url to retrieve shadowsocks configuration: {url}")
    ss_encoded_url = get_ss_url(url)
    launch_outline_cli(ss_encoded_url)