AirlockTools/utils/allowlist.py

#    Copyright (C) 2025 James Brotosky, Brandon Wickline
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as published
#    by the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
import datetime
import requests
import json
import os
import pandas
import time
import utils.pretty as ct
import ijson
import os
from bson import ObjectId
import datetime

def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): 
    file_path = 'chunkinator.json'
    headers = {
        "X-APIKey": os.getenv('APIKEY')
    }
    checkpoint = '000000000000000000000000'
    json_output = {'error': 'Success', 'response': {'exechistories': []}}

    while True:
        json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers)
        histories = json_response_data['response']['exechistories']
        if not histories:
            break

        array_dividend = max(round(len(histories) / 20), 1)
        match_found = False
        array_dividend = round(len(json_response_data['response']['exechistories'])/20)
        if array_dividend == 0:
            array_dividend == 1
        for index, item in enumerate(json_response_data['response']['exechistories'][::array_dividend]):
            if (datetime.date.today() - datetime.timedelta(days=60) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
                match_found = True
                break
        checkpoints_processed = round(len(json_response_data['response']['exechistories'])/array_dividend)
        print(ct.colorText(f"{checkpoints_processed} checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
        checkpoint = item['checkpoint']
        if match_found == True:
            for index, item in enumerate(json_response_data['response']['exechistories']):
                if index == len(json_response_data['response']['exechistories']) -1:
                    checkpoint = item['checkpoint']
                    print(ct.colorText(f"All Events Processed. Next Checkpoint: {item['checkpoint']}", "blue"))
                    break
                else:
                    if (datetime.date.today() - datetime.timedelta(days=60) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
                        pass
                    else:
                        #print(f"Added: {item['datetime']} | {item['checkpoint']} | {item['filename']} | {item['sha256']}")
                        json_output['response']['exechistories'].append(item)
            if os.path.exists(file_path):
                seen = {}
                with open(file_path, 'r') as file:
                    temp_file = json.load(file)
                combined_data =  temp_file['response']['exechistories'] + json_output['response']['exechistories']
                for item in combined_data:
                    key = (item.get('sha256'), item.get('filename'), item.get('hostname'))
                    seen[key] = item
                deduplicated_data = list(seen.values())
                with open(file_path, 'w') as file:
                    json.dump({'error': 'Success', 'response': {'exechistories': deduplicated_data}}, file)
                    json_output = {'error': 'Success', 'response': {'exechistories': []}}
            else:
                with open(file_path, 'a') as file:
                    json.dump(json_output, file)
                    json_output = {'error': 'Success', 'response': {'exechistories': []}}

    with open(file_path, 'r') as file:
        json_output = json.load(file)
    json_output = json.dumps(json_output)
    os.remove(file_path)
    if outputjson == True:
        return json_output

def checkpoint_stomper(checkpoint, url, policy, headers):
    json_output = {'error': 'Success', 'response': {'exechistories': []}}
    endpoint = url + '/v1/logging/exechistories'
    payload_dict = {
        "type":[1,2,6,7],
        "checkpoint": checkpoint,
        "policy": [policy]
    }
    payload = json.dumps(payload_dict)
    with requests.request("POST", endpoint, headers=headers, data=payload, verify=False, stream=True) as response:
        parser = ijson.items(response.raw, 'response.exechistories.item')
        for item in parser:
            key = (item.get('sha256'), item.get('hostname'))
            if key not in json_output:
                json_output['response']['exechistories'].append(item)   
    parse_text = json.loads(json.dumps(json_output))
    return parse_text

def listPolicies(url): 
    endpoint = url + '/v1/group'
    print(ct.colorText("[+] Grabbing All Policies", "cyan"))
    payload = {}
    headers = {
        "X-APIKey": os.getenv('APIKEY')
    }
    response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
    parse_text = json.loads(response.text)
    policiesnames = []
    policyids = []
    for index, list in enumerate(parse_text['response']['groups'], start=1):
        print(ct.colorText(f"{index}. {list['name']}", "yellow"))
        policiesnames.append(list['name'])
        policyids.append(list['groupid'])
    choice = input(ct.colorText("Select Policy Group: ", "white"))
    choice = int(choice) - 1
    return choice, policiesnames, policyids

def listAllowlists(url): 
    endpoint = url + '/v1/application'
    print(ct.colorText("[+] Grabbing All Allowlists", "cyan"))
    payload = {}
    headers = {
        "X-APIKey": os.getenv('APIKEY')
    }
    response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
    parse_text = json.loads(response.text)
    policiesnames = []
    policyids = []
    for index, list in enumerate(parse_text['response']['applications'], start=1):
        if index >= 38:
            print(ct.colorText(f"{index}. {list['name']}", "yellow"))
            policiesnames.append(list['name'])
            policyids.append(list['applicationid'])
    choice = int(input(ct.colorText("Select allowlist: ", "white")))
    if choice < 38:
        print(ct.colorText("Please only choose an allowlist designed for this use - '38+'","red"))
    elif choice >= 38:
        choice = choice - 38
        return choice, policiesnames, policyids
    #Need else and catch for upper bound

def skipback(days):
    """
    Generate a MongoDB ObjectId for a given number of days ago from today.
    Adds 1 extra day to the input to look further back.
    """
    adjusted_days = days + 1
    date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=adjusted_days)
    timestamp = int(date_days_ago.timestamp())
    hex_timestamp = format(timestamp, '08x')
    objectid_hex = hex_timestamp + '0000000000000000'
    return ObjectId(objectid_hex)