import logging import json import hashlib import requests import time import math from collections import OrderedDict from datetime import datetime import os import socket import traceback import subprocess import tempfile from subprocess import PIPE # Convenient constants - Do these better later config_path = '/etc/goliath_agent' out_path = '/etc/goliath/appliance/config' class pyCreatia: #Constants CREATIA_AUTH_ISSUE = 'auth/issue/' CREATIA_AUTH_REVOKE = 'auth/revoke/' def __init__(self, endpoint: str, identify_token: str, secret_token: str): self.creatia_logger = logging.getLogger("creatia_api") self.identify_token = identify_token self.secret_token = secret_token self.creatia_bearer_token = None self.creatia_bearer_expires = 0 self.endpoint = endpoint def is_token_expired(self): return (float(datetime.now().timestamp()) > self.creatia_bearer_expires) def creatia_authenticate(self) -> bool: self.creatia_logger.info("Authenicating") if self.is_token_expired(): self.creatia_logger.info("Expired or empty token") response = requests.post(self.endpoint + self.CREATIA_AUTH_ISSUE, headers=self.creatia_construct_headers(), data={ 'identifier': self.identify_token, 'token': self.secret_token }, timeout=15) if response.ok: self.creatia_logger.info("Valid authentication attempt") self.creatia_bearer_token = response.json()["bearer"] self.creatia_bearer_expires = response.json()["expiry"] return True self.creatia_logger.error(f"Failed authetication - {response.status_code} - {response.text}") raise Exception("Failed authetication") def creatia_construct_headers(self) -> dict: headers = { 'User-Agent': f'pyCreatia-Goliath' } if self.creatia_bearer_token is not None and not self.is_token_expired(): headers['Authorization'] = 'Bearer ' + self.creatia_bearer_token return headers def creatia_revoke(self) -> bool: self.creatia_authenticate() self.creatia_logger.info("Revoking") if self.creatia_bearer_token is not None: response = requests.post(self.endpoint + self.CREATIA_AUTH_REVOKE, headers=self.creatia_construct_headers(), timeout=15) if response.status_code == 200: self.creatia_logger.info("Token revoked") self.creatia_bearer_token = None self.creatia_bearer_expires = 0 return True self.creatia_logger.error(f"Failed token revoke - {response.status_code} - {response.text}") raise Exception("Failed token revoke") def creatia_get(self, endpoint: str, in_params: dict) -> dict: self.creatia_authenticate() self.creatia_logger.debug(f"GET REQUEST: {endpoint} - {in_params}") response = requests.get(self.endpoint + endpoint, params=in_params, headers=self.creatia_construct_headers(), timeout=15) if response.status_code == 200: self.creatia_logger.debug(f"GET RESPONSE: {response.status_code} - {response.text}") return response.json() raise Exception("GET invalid", response.status_code, response.text) def creatia_post(self, endpoint: str, payload: dict, files: dict = {}) -> dict: self.creatia_authenticate() response = requests.post(self.endpoint + endpoint, json=payload, headers=self.creatia_construct_headers(), files=files, timeout=15) if response.status_code == 200: return response.json() raise Exception("POST invalid", response.status_code, response.text) def creatia_patch(self, endpoint: str, payload: dict) -> dict: self.creatia_authenticate() response = requests.patch(self.endpoint + endpoint, json=payload, headers=self.creatia_construct_headers(), timeout=15) if response.status_code == 200: return response.json() raise Exception("PATCH invalid", response.status_code, response.text) class Agent: def __init__(self, config): # Store logger self.logger = logging.getLogger("creatia_agent") # Recall full configuration self.full_config = config # Fetch credentials and important information from configuration.json self.identity = self.full_config['CREATIA_IDENTITY'] self.secret = self.full_config['CREATIA_SECRET'] self.region = self.full_config['CREATIA_REGION'] self.subregion = self.full_config['CREATIA_SUBREGION'] self.facility = self.full_config['CREATIA_FACILITY'] self.appliance_id = self.full_config['CREATIA_APPLIANCE_ID'] self.level = 1 if 'AGENT_LEVEL' not in self.full_config else self.full_config['AGENT_LEVEL'] # Setup API endpoint self.api = pyCreatia(self.full_config['CREATIA_ENDPOINT'], self.identity, self.secret) self.fingerprint_to_uuid = {} def safe_write(self, filepath: str, content: str) -> bool: """Write to a temp file in the same directory, then atomically replace the target.""" dir_name = os.path.dirname(filepath) try: # Use a temp file in the same directory to ensure same filesystem (required for os.replace) fd, tmp_path = tempfile.mkstemp(dir=dir_name, prefix='.tmp_') try: with os.fdopen(fd, 'w') as f: f.write(content) f.flush() os.fsync(f.fileno()) # Flush kernel buffers to disk os.replace(tmp_path, filepath) # Atomic on POSIX systems return True except Exception: os.unlink(tmp_path) # Clean up temp file on failure raise except IOError as e: self.logger.error(f"Safe write failed for {filepath}: {str(e)}") return False def run_command(self, wd: str, values: list): full_command = [] for value in values: full_command = full_command + [value.strip()] command = subprocess.run(full_command, stdout=PIPE, stderr=PIPE, encoding='utf-8', cwd=wd) if command.returncode != 0: return False return command.stdout def run_update_hash(self, chains, targets): # Get the hash for the running files chain_hash = hashlib.md5(chains.encode('utf-8')).hexdigest() target_hash = hashlib.md5(targets.encode('utf-8')).hexdigest() running_hash = f'{chain_hash}|{target_hash}' # Get the hash for the agent agent_hash = hashlib.md5(' '.join(list(open('/usr/bin/goliath_agent/agent.py'))).encode('utf-8')).hexdigest() # Get the hash for the DPDK appliance appliance_dir_slug = '' if os.path.exists('/etc/goliath-new/appliance/'): appliance_dir_slug = 'goliath-new' elif os.path.exists('/etc/goliath/appliance/'): appliance_dir_slug = 'goliath' goliath_commit = 'Unobtainable' if appliance_dir_slug != '': command_output = self.run_command(f'/etc/{appliance_dir_slug}/appliance/', [ 'git', 'rev-parse', 'HEAD' ]) if command_output is not False: goliath_commit = command_output self.api.creatia_post('goliath/hash/', { 'appliance_id': self.appliance_id, 'hash': running_hash, 'agent_commit': agent_hash, 'goliath_commit': goliath_commit }) self.api.creatia_revoke() def run_identify_appliance(self): # Whenever an appliance registers itself, OR whenever it restarts, we need to make sure Creatia is aware of it's existence. payload = { 'name': socket.gethostname(), 'region_tag': self.region, 'subregion_tag': self.subregion, 'facility_tag': self.facility } if self.appliance_id != '': payload['appliance_id'] = self.appliance_id result = self.api.creatia_post('goliath/register/', payload) self.api.creatia_revoke() if(result['message'] == 'success' and self.appliance_id == ''): self.appliance_id = str(result['appliance_id']) self.full_config['CREATIA_APPLIANCE_ID'] = self.appliance_id with open(config_path + '/' + 'goliath_agent.conf', 'w') as new_config_file: try: new_config_file.write(json.dumps(self.full_config, indent=2)) except IOError as e: self.logger.error(f"Failed to write to configuration file to recall appliance id. Exception has occured - {str(e)}") def run_fetch_update(self, stored_chain_hash, stored_target_hash, stored_count): # Stuff to control the flow update_configurations = False # Fetch stuff try: # We should check if the file themselves are accurate - If they're not, then obviously we should rewrite them. running_chain_hash = 'NO CHAIN FILE WRITTEN' running_targets_hash = 'NO TARGET FILE WRITTEN' with open(f'{out_path}/chains.json', 'r') as f: content = f.read() running_chain_hash = hashlib.md5(content.encode()).hexdigest() with open(f'{out_path}/targets.json', 'r') as f: content = f.read() running_targets_hash = hashlib.md5(content.encode()).hexdigest() logger.warning(f'Chain hashes. Running: {running_chain_hash}, stored: {stored_chain_hash}') logger.warning(f'Target hashes. Running: {running_targets_hash}, stored: {stored_target_hash}') # Region specific configuration pulling is a juicy one indeed raw_information = self.api.creatia_get('goliath/data-v4', { 'appliance_id': self.appliance_id, 'region': self.region, 'subregion': self.subregion, 'datacentre': self.facility, 'chain_hash': running_chain_hash, 'target_hash': running_targets_hash }) self.api.creatia_revoke() except Exception as e: self.logger.error(f"Could not read a 200 from Creatia - {str(e)}") return stored_chain_hash, stored_target_hash, stored_count # Determine if it's valid - Write if update is true AND: # - The update is not empty # - There hasn't been a sharp reduction in the number of prefixes (70% of previous count of prefixes) if raw_information['update'] == True: update_configurations = True if (raw_information['prefix_count'] == 0) or ((raw_information['prefix_count'] / stored_count) <= 0.7): logger.warning(f"Concern - No prefixes provided or reduction to 70% of previous count") update_configurations = False if raw_information['targets'] == '' or raw_information['targets'] == '[]' or raw_information['targets'] == '{}' or raw_information['targets'] == '[{}]' : logger.warning(f"Concern - Empty file provided for targets") update_configurations = False # Write if valid if update_configurations: os.makedirs(out_path, exist_ok=True) # Write chain if not self.safe_write(f'{out_path}/chains.json', raw_information['chains']): self.logger.error("Failed to safe-write chains file, skipping update") return stored_chain_hash, stored_target_hash, stored_count logger.warning("Wrote to chains file") # Write targets if not self.safe_write(f'{out_path}/targets.json', raw_information['targets']): self.logger.error("Failed to safe-write targets file, skipping update") return stored_chain_hash, stored_target_hash, stored_count logger.warning("Wrote to targets file") self.run_update_hash(raw_information['chains'], raw_information['targets']) return raw_information['chain_hash'], raw_information['target_hash'], raw_information['prefix_count'] def run_track_attacks(self, known_attacks): try: measuring_at = datetime.now().timestamp() - 60 # Clean out old attacks - Anything that hasn't been seen in five minutes... for key in list(known_attacks): if measuring_at - known_attacks[key]['whole_last_timestamp'] > 300: del known_attacks[key] # Add in new attacks and reserve to send attacks = requests.get('http://127.0.0.1:5000/events', timeout=15).json() for attack in attacks: seconds_timestamp = time.mktime(datetime.strptime(attack['timestamp'], "%Y-%m-%dT%H:%M:%SZ").timetuple()) if f'''{attack['description']}/{attack['dst_addr']}''' not in known_attacks: known_attacks[f'''{attack['description']}/{attack['dst_addr']}'''] = { 'rounded_start_timestamp': math.floor(seconds_timestamp/60)*60, 'whole_start_timestamp': measuring_at, 'rounded_last_timestamp': math.ceil(seconds_timestamp/60)*60, 'whole_last_timestamp': measuring_at, 'type': attack ['type'], 'destination': attack['dst_addr'], 'description': attack['description'] } else: known_attacks[f'''{attack['description']}/{attack['dst_addr']}''']['rounded_last_timestamp'] = math.ceil(seconds_timestamp/60)*60 known_attacks[f'''{attack['description']}/{attack['dst_addr']}''']['whole_last_timestamp'] = measuring_at self.api.creatia_post('goliath/attack/', { 'appliance_id': self.appliance_id, 'attacks': known_attacks }) except Exception as e: logger.error(e) logger.error(f"Attack logging not running") return known_attacks def cycle(self, stored_chain_hash, stored_target_hash, stored_count, stored_attacks): self.run_identify_appliance() stored_chain_hash, stored_target_hash, stored_count = self.run_fetch_update(stored_chain_hash, stored_target_hash, stored_count) stored_attacks = self.run_track_attacks(stored_attacks) return stored_chain_hash, stored_target_hash, stored_count, stored_attacks if __name__ == "__main__": # Logger logger = logging.getLogger("main") # Initialise the configuration config_options = { 'CREATIA_IDENTITY': '', 'CREATIA_SECRET': '', 'CREATIA_REGION': '', 'CREATIA_SUBREGION': '', 'CREATIA_FACILITY': '', 'CREATIA_APPLIANCE_ID': '', 'CREATIA_ENDPOINT': 'https://api-creatia.securelayer.com/api/', 'AGENT_SLEEP_TIME': 10, 'AGENT_LEVEL': 1 } if os.path.exists(config_path + '/' + 'goliath_agent.conf'): config_options = json.load(open(config_path + '/' + 'goliath_agent.conf')) else: os.makedirs(config_path, exist_ok=True) with open(config_path + '/' + 'goliath_agent.conf', 'w') as new_config_file: try: new_config_file.write(json.dumps(config_options, indent=2)) except IOError as e: logger.error(f"Failed to write new configuration file - {str(e)}") # Anything we need to recall between loops stored_attacks = {} stored_chain_hash = '' stored_target_hash = '' stored_count = 1 # Begin the programs main loop while True: try: agent = Agent( config_options ) # Lifecycle and process outputs from the lifecycle that must be retained. stored_chain_hash, stored_target_hash, stored_count, stored_attacks = agent.cycle(stored_chain_hash, stored_target_hash, stored_count, stored_attacks) except Exception as e: logger.error(f"Failed to run agent process method. Exception has occured - {str(e)}") logger.error(traceback.format_exc()) time.sleep(config_options['AGENT_SLEEP_TIME'])