#!/usr/bin/env python3 # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. # -*- coding: utf-8 -*- # This script serves the purpose to replace sensitive parts in the jenkins configuration with placeholders in order # to allow the configuration to be published to a public repository import argparse import filecmp import glob import json import logging import os import pathlib import shutil from collections import namedtuple from lxml import etree SECRET_ENTRY_KEYS = ['filepath', 'xpath', 'secret', 'placeholder'] SecretEntry = namedtuple('SecretEntry', SECRET_ENTRY_KEYS) SYMLINK_ENTRY_KEYS = ['filepath', 'is_dir'] SymlinkEntry = namedtuple('SymlinkEntry', SYMLINK_ENTRY_KEYS) def main(): parser = argparse.ArgumentParser() parser.add_argument('-vf', '--varfile', help='Location of the variable file', type=str) parser.add_argument('-sf', '--symlinkfile', help='Location of the symlink file', type=str) parser.add_argument('-sd', '--secretsdir', help='Location of the directory containing secrets', type=str) parser.add_argument('-jd', '--jenkinsdir', help='Location of the jenkins directory', type=str) parser.add_argument('-m', '--mode', help='"remove" or "insert" credentials', default='insert', type=str) args = parser.parse_args() execute_config_templating(args.varfile, args.secretsdir, args.jenkinsdir, args.mode, update_secrets=False) # TODO: Add symlink list creation def execute_config_templating(varfile, secretsdir, jenkinsdir, mode, update_secrets): """ Execute config templating that inserts or removes secrets from a jenkins configuration directory :param varfile: File containing the actual variables that should be used during replacement :param secretsdir: Directory containing all files that should be just copied or removed as secrets :param jenkinsdir: Jenkins configuration directory :param mode: 'insert' replaces placeholders with actual values. 'remove' removes these values and inserts placeholders :param update_secrets: Update secrets if mode == 'remove' and actual secrets differ from the stored ones :return: """ secret_entries = read_secret_entires(varfile) logging.debug('Found {} secret entries to be replaced'.format(len(secret_entries))) # Prepare by finding all unique identifiers. This is required because XML parsers do not allow in-place # replacements. Instead, we're reading a unique identifier under the specified xpath and then try to # replace it in-place by using search-and-replace. This method will be aborted if the current value # has been found multiple times within the same file. for root, dirs, files in os.walk(secretsdir, topdown=False): for name in files: original_path = os.path.join(root, name) rel_path = os.path.relpath(original_path, secretsdir) temp_path = os.path.join(jenkinsdir, rel_path) if mode == 'insert': pathlib.Path(os.path.dirname(temp_path)).mkdir(parents=True, exist_ok=True) shutil.copyfile(original_path, temp_path) elif mode == 'remove': # Check if secret does not exist anymore if not os.path.isfile(temp_path): if update_secrets: logging.info('Deleting secret {} because it has been removed on target'.format(rel_path)) os.remove(original_path) else: raise ValueError('Secret {} has been deleted'.format(rel_path)) # Check if secrets are the same or have to be updated if not filecmp.cmp(temp_path, original_path): if update_secrets: logging.info('Replacing secret {} due to changed content'.format(rel_path)) shutil.copyfile(temp_path, original_path) else: raise ValueError('Secret {} contains changed content'.format(rel_path)) os.remove(temp_path) else: raise ValueError('Mode {} unknown'.format(mode)) # TODO check this previously # Check if any files in the secrets-dir are left that didn't exist in the previous config. Unfortunately, # we can't verify if secrets outside the secrets-dir have been added. if mode == 'remove': temp_secrets_dir = os.path.join(jenkinsdir, 'secrets') for root, dirs, files in os.walk(temp_secrets_dir, topdown=False): for name in files: temp_path = os.path.join(root, name) rel_path = os.path.relpath(temp_path, jenkinsdir) original_path = os.path.join(secretsdir, rel_path) if update_secrets: logging.info('Adding new secret at {}'.format(rel_path)) shutil.copyfile(temp_path, original_path) else: raise ValueError('New secret at {}'.format(rel_path)) shutil.rmtree(temp_secrets_dir) for secret_entry in secret_entries: temp_file_path = os.path.join(jenkinsdir, secret_entry.filepath) if os.path.isfile(temp_file_path): element = etree.parse(temp_file_path).xpath(secret_entry.xpath) # Check if xpath delivers multiple results. The xpath should only match once if len(element) == 1: current_value = element[0].text if not current_value.strip(): raise ValueError('Element {} at {}:{} is not a text field'. format(current_value, temp_file_path, secret_entry.xpath)) if mode == 'insert': expected_value = secret_entry.placeholder target_value = secret_entry.secret elif mode == 'remove': expected_value = secret_entry.secret target_value = secret_entry.placeholder else: raise ValueError('Mode {} unknown'.format(mode)) if current_value == expected_value: logging.debug( 'Replacing {} with {} at {}:{}'.format(current_value, target_value, secret_entry.xpath, temp_file_path)) _replace_values(current_value, target_value, temp_file_path) elif current_value == target_value: logging.info('Target value {} already present. Skipping {}:{}'. format(target_value, secret_entry.xpath, temp_file_path)) continue else: raise ValueError('Current value "{}" does not match expected value "{}" in {}:{}'.format( current_value, expected_value, secret_entry.xpath)) elif len(element) == 0: raise ValueError('Element at {}:{} not found'.format(temp_file_path, secret_entry.xpath)) else: raise ValueError('1 Element expected, {} found at {}:{}'. format(len(element), temp_file_path, secret_entry.xpath)) else: raise FileNotFoundError('Could not find file {}'.format(temp_file_path)) def assemble_symlink_list(symlink_file, jenkinsdir): """ Assemble a list of files that should be symlinked during startup, providing support for state files on EBS :param symlink_file: File containing path expressions to describe the symlinked files and dirs :param jenkinsdir: Jenkins configuration directory :return: Array of SymlinkEntry """ symlink_config = read_symlink_entries(symlink_file) logging.debug('Found {} symlink entries'.format(len(symlink_config))) symlinks = [] for symlink_entry in symlink_config: input_path_expression = os.path.join(jenkinsdir, symlink_entry.filepath) result_paths = [] if '*' in symlink_entry.filepath: # If path contains a wildcard, search for all results wildcard_path_split = input_path_expression.split('*') if len(wildcard_path_split) == 1: result_paths = glob.glob(input_path_expression) elif len(wildcard_path_split) == 2: # If wildcard is in the middle and target directories don't exist, result would be empty. # Instead, iterate manually for partial_path in glob.glob(os.path.join(wildcard_path_split[0], '*')): result_paths.append(os.path.join(partial_path, wildcard_path_split[1].lstrip('/'))) else: raise ValueError('Symlink expression may only contain one wildcard') logging.debug('Resolving {} to {}'.format(symlink_entry.filepath, result_paths)) else: result_paths = [input_path_expression] for abs_path in result_paths: rel_path = os.path.relpath(abs_path, jenkinsdir) symlinks.append(SymlinkEntry(rel_path, symlink_entry.is_dir)) return symlinks def read_secret_entires(varfile): """ Read SecretEntry from varfile :param varfile: File containing SecretEntries as JSON :return: Array of SecretEntry """ with open(varfile, 'r') as fp: secret_dict_raw = json.load(fp) secrets = [] for secret_entry_dict in secret_dict_raw: # Import values in the same order as expected in SecretEntry. Reason being that the values are expected # to be inserted in the same order as defined previously. secrets.append(SecretEntry(*[secret_entry_dict[k] for k in SECRET_ENTRY_KEYS])) return secrets def read_symlink_entries(varfile): """ Read SymlinkEntry from varfile :param varfile: File containing SymlinkEntries as JSON :return: Array of SymlinkEntries """ with open(varfile, 'r') as fp: symlink_dict_raw = json.load(fp) symlinks = [] for symlink_entry_dict in symlink_dict_raw: # Import values in the same order as expected in SecretEntry. Reason being that the values are expected # to be inserted in the same order as defined previously. symlinks.append(SymlinkEntry(*[symlink_entry_dict[k] for k in SYMLINK_ENTRY_KEYS])) return symlinks def _replace_values(current_value, target_value, temp_file_path): # Only execute changes on the temp file with open(temp_file_path, 'r+') as temp_fh: temp_file_content = temp_fh.read() # Check if current value is unique within the temp file to prevent injection attacks nb_occurrences_temp = temp_file_content.count(current_value) if nb_occurrences_temp != 1: raise ValueError( '1 occurrence of current value {} in file {} expected, {} found'.format( current_value, temp_file_path, nb_occurrences_temp)) temp_file_content_replaced = temp_file_content.replace(current_value, target_value) # Write replaced content back to the temp file temp_fh.seek(0) temp_fh.write(temp_file_content_replaced) temp_fh.truncate() if __name__ == '__main__': main()