auth_remote_user

#!/usr/bin/env python3
'''
Ext filter that will validate POST requests and make sure REMOTE_USER is used
as username. This enables to use Saltstack's sharedsecret auth module and we
can use technologies as Kerberos GSSAPI to authenticate an API user to
Saltstack. For more information, visit:
https://github.com/stockholmuniversity/salt-netapi-remoteuser-auth
'''

import argparse
import json
import os
import re
import sys
import unittest
import urllib.parse

# pylint: disable=import-error
import yaml
# pylint: disable=unused-import
from su.logging import logging, structured  # noqa: F401


def update_indata_dict(indata_dict, remote_user, password, eauth):
    ''' Always update username to REMOTE_USER. This will prevent a user to
        tamper with the username parameter.
    '''
    indata_dict.update(username=remote_user)
    indata_dict.update(password=password)
    indata_dict.update(eauth=eauth)


def update_indata_lowstate(indata, remote_user, password, eauth):
    ''' When using JSON for the lowstate you can send multiple lowstates as a
    list, so we need to handle that.
    '''
    if isinstance(indata, list):
        for lowstate in indata:
            update_indata_dict(lowstate, remote_user, password, eauth)
    else:
        update_indata_dict(indata, remote_user, password, eauth)


def main(stdin):
    ''' Take data on standard in and make sure we are setting username to
        REMOTE_USER.
    '''

    logger = logging.getLogger("auth_remote_user")
    logger.setLevel(logging.INFO)

    remote_user = os.getenv("REMOTE_USER")
    if remote_user is None:
        logger.error("REMOTE_USER isn't set, check your configuration.")
        sys.exit(1)

    remote_user = authz_regex(username=remote_user)
    password = None
    eauth = os.environ.get('EAUTH', 'sharedsecret')

    if eauth == 'sharedsecret':
        shared_secret_config = os.environ.get('SHARED_SECRET_CONFIG')
        if shared_secret_config is None:
            logger.error(
                "Can't find required environment variables: {env}".format(
                    env={
                        "remote_user": remote_user,
                        "shared_secret_config": shared_secret_config,
                    }))
            sys.exit(1)

        yaml_data = {}
        try:
            with open(shared_secret_config, 'r') as stream:
                yaml_data = yaml.safe_load(stream)
        # pylint: disable=invalid-name
        except (yaml.YAMLError, FileNotFoundError) as e:
            logger.exception(e)
            sys.exit(1)

        password = yaml_data.get('sharedsecret')
        if password is None:
            logger.error("Can't find sharedsecret in config {}".format(
                shared_secret_config))
            sys.exit(1)

    indata = stdin.read()

    indata_dict = {}
    request_parameters = ""
    if os.environ.get('CONTENT_TYPE') == 'application/x-www-form-urlencoded':
        logger.debug("Handling a {} request".format(
            os.environ.get('CONTENT_TYPE')))
        indata_dict = urllib.parse.parse_qs(indata)
        update_indata_lowstate(indata_dict, remote_user, password, eauth)
        request_parameters = urllib.parse.urlencode(indata_dict, doseq=True)
    elif os.environ.get('CONTENT_TYPE') == 'application/json':
        logger.debug("Handling a {} request".format(
            os.environ.get('CONTENT_TYPE')))
        indata_dict = None
        try:
            indata_dict = json.loads(indata)
        # pylint: disable=invalid-name
        except json.decoder.JSONDecodeError as e:
            logger.exception(e)
            sys.exit(1)
        update_indata_lowstate(indata_dict, remote_user, password, eauth)
        request_parameters = json.dumps(indata_dict)
    else:
        logger.error("Unknown request with content_type={}".format(
            os.environ.get('CONTENT_TYPE')))
        sys.exit(1)

    print(request_parameters)


def authz_regex(username=os.getenv("REMOTE_USER"),
                pattern=os.getenv("REMOTE_USER_PATTERN", "")):
    """
    Authorizes a username based on a regex.

    Also provides the option to modify the username via a capture group. The
    first group is used as the username.
    """
    if username is None:
        return None
    match = re.match(pattern, username)
    # The regex matched grouped or not
    if match is not None:
        matched_username = None
        try:
            matched_username = match.group(1)
        except IndexError:
            return username
        return matched_username
    return None


class SaltAuthZTest(unittest.TestCase):
    # pylint: disable=missing-docstring,anomalous-backslash-in-string

    remote_user_pattern = '^(.+?)\/root@SU\.SE$'  # noqa: W605

    def test_root(self):
        self.assertEqual(
            authz_regex(username="simlu/root@SU.SE",
                        pattern=self.remote_user_pattern), "simlu")

    def test_hackerman(self):
        self.assertEqual(
            authz_regex(username="/root@SU.SE",
                        pattern=self.remote_user_pattern), None)

    def test_a(self):
        self.assertEqual(
            authz_regex(username="a/root@SU.SE",
                        pattern=self.remote_user_pattern), "a")

    def test_admin(self):
        self.assertEqual(
            authz_regex(username="simlu/admin@SU.SE",
                        pattern=self.remote_user_pattern), None)

    def test_regular(self):
        self.assertEqual(
            authz_regex(username="simlu@SU.SE",
                        pattern=self.remote_user_pattern), None)

    def test_regular_no_realm(self):
        self.assertEqual(
            authz_regex(username="simlu", pattern=self.remote_user_pattern),
            None)

    def test_regular_no_pattern(self):
        self.assertEqual(authz_regex(username="simlu@SU.SE"), "simlu@SU.SE")

    def test_realm_pattern_allowed(self):
        self.assertEqual(
            authz_regex(username="simlu@SU.SE", pattern=r".*@SU\.SE$"),
            "simlu@SU.SE")

    def test_realm_pattern_denied(self):
        self.assertEqual(
            authz_regex(username="simlu@SOY.SE", pattern=r".*@SU\.SE$"), None)

    def test_nada(self):
        self.assertEqual(authz_regex(), None)

    def test_none(self):
        self.assertEqual(authz_regex(username=None), None)

    def test_none_everything(self):
        self.assertEqual(authz_regex(username=None, pattern=None), None)


class SaltAuthZMultiUserTest(unittest.TestCase):
    # pylint: disable=missing-docstring,anomalous-backslash-in-string

    remote_user_pattern = r'^(su-.+?|.+?(?=\/root))(?:\/root)?@SU\.SE$'

    def test_regular(self):
        self.assertEqual(
            authz_regex(username="simlu@SU.SE",
                        pattern=self.remote_user_pattern), None)

    def test_serviceaccount(self):
        self.assertEqual(
            authz_regex(username="su-ci-prod@SU.SE",
                        pattern=self.remote_user_pattern), 'su-ci-prod')

    def test_root(self):
        self.assertEqual(
            authz_regex(username="simlu/root@SU.SE",
                        pattern=self.remote_user_pattern), 'simlu')

    def test_ppp(self):
        self.assertEqual(
            authz_regex(username="simlu/ppp@SU.SE",
                        pattern=self.remote_user_pattern), None)


if __name__ == "__main__":
    PARSER = argparse.ArgumentParser(
        description='Authenticate REMOTE_USER for Salt NetAPI')
    PARSER.add_argument('--test', action='store_true', help='Run unit tests')
    ARGS = PARSER.parse_args()
    if ARGS.test:
        unittest.main(argv=[sys.argv[0]])
    else:
        main(sys.stdin)