Function lambda_handler_user

bless/aws_lambda/bless_lambda_user.py:34–193

This is the function that will be called when the lambda function starts. :param event: Dictionary of the json request. :param context: AWS LambdaContext Object http://docs.aws.amazon.com/lambda/latest/dg/python-context-object.html :param ca_private_key_password: For local testi

(
        event, context=None, ca_private_key_password=None,
        entropy_check=True,
        config_file=None)

def lambda_handler_user(
        event, context=None, ca_private_key_password=None,
        entropy_check=True,
        config_file=None):
    """
    This is the function that will be called when the lambda function starts.
    :param event: Dictionary of the json request.
    :param context: AWS LambdaContext Object
    http://docs.aws.amazon.com/lambda/latest/dg/python-context-object.html
    :param ca_private_key_password: For local testing, if the password is provided, skip the KMS
    decrypt.
    :param entropy_check: For local testing, if set to false, it will skip checking entropy and
    won't try to fetch additional random from KMS.
    :param config_file: The config file to load the SSH CA private key from, and additional settings.
    :return: the SSH Certificate that can be written to id_rsa-cert.pub or similar file.
    """
    bless_cache = setup_lambda_cache(ca_private_key_password, config_file)

    # AWS Region determines configs related to KMS
    region = bless_cache.region

    # Load the deployment config values
    config = bless_cache.config

    logger = set_logger(config)

    certificate_validity_before_seconds = config.getint(BLESS_OPTIONS_SECTION,
                                                        CERTIFICATE_VALIDITY_BEFORE_SEC_OPTION)
    certificate_validity_after_seconds = config.getint(BLESS_OPTIONS_SECTION,
                                                       CERTIFICATE_VALIDITY_AFTER_SEC_OPTION)
    ca_private_key = config.getprivatekey()
    certificate_extensions = config.get(BLESS_OPTIONS_SECTION, CERTIFICATE_EXTENSIONS_OPTION)

    # Process cert request
    schema = BlessUserSchema(strict=True)
    schema.context[USERNAME_VALIDATION_OPTION] = config.get(BLESS_OPTIONS_SECTION, USERNAME_VALIDATION_OPTION)
    schema.context[REMOTE_USERNAMES_VALIDATION_OPTION] = config.get(BLESS_OPTIONS_SECTION,
                                                                    REMOTE_USERNAMES_VALIDATION_OPTION)
    schema.context[REMOTE_USERNAMES_BLACKLIST_OPTION] = config.get(BLESS_OPTIONS_SECTION,
                                                                   REMOTE_USERNAMES_BLACKLIST_OPTION)

    try:
        request = schema.load(event).data
    except ValidationError as e:
        return error_response('InputValidationError', str(e))

    logger.info('Bless lambda invoked by [user: {0}, bastion_ips:{1}, public_key: {2}, kmsauth_token:{3}]'.format(
        request.bastion_user,
        request.bastion_user_ip,
        request.public_key_to_sign,
        request.kmsauth_token))

    # Make sure we have the ca private key password
    if bless_cache.ca_private_key_password is None:
        return error_response('ClientError', bless_cache.ca_private_key_password_error)
    else:
        ca_private_key_password = bless_cache.ca_private_key_password