AirlockTools/utils/clientfunctions.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/>.

#Local Imports
import utils.pretty as ct

#Standard Libary Imports:
import datetime
import json
import os

#3rd Party Imports:
import pandas as pd
import requests

def findAgentID(url):
	   
    print(ct.colorText("WARNING: Device Name is Case Sensitive", "red"))
    hostname = input(ct.colorText("Enter Device Name: ", "white"))
    
    endpoint = url + '/v1/agent/find'
    payload = {
            "hostname" : f"{hostname}"
    }

    headers = {"X-APIKey": os.getenv('APIKEY')}
    payload = json.dumps(payload)


    response = requests.post(endpoint, headers=headers, data=payload, verify=False)
    result = json.loads(response.text)
    data = pd.DataFrame(result["response"]["agents"])
    return data.loc[0, "agentid"]

def getDestAllowlistFromClientID(url, clientid):

    #This is dependant on their being an allowlist in the policy containing the name "localapproval"

    endpoint = url + '/v1/agent/find'
    payload = {
            "agentid" : f"{clientid}"
    }

    headers = {"X-APIKey": os.getenv('APIKEY')}
    payload = json.dumps(payload)


    response = requests.post(endpoint, headers=headers, data=payload, verify=False)
    result = json.loads(response.text)
    data = pd.DataFrame(result["response"]["agents"])

    allowlists = getPolicyAllowlists(url,data.loc[0, "groupid"]) 
 
    matches = allowlists.loc[
        allowlists['name'].str.contains('local', case=False, na=False) &
        allowlists['name'].str.contains('approval', case=False, na=False),
        'applicationid'
    ].values
    app_id = matches[0] if len(matches) > 0 else None
    
    return app_id

def getPolicyFromClientID(url, clientid):

    endpoint = url + '/v1/agent/find'
    payload = {
            "agentid" : f"{clientid}"
    }

    headers = {"X-APIKey": os.getenv('APIKEY')}
    payload = json.dumps(payload)


    response = requests.post(endpoint, headers=headers, data=payload, verify=False)
    result = json.loads(response.text)
    data = pd.DataFrame(result["response"]["agents"])

    policy = getPolicyName(url,data.loc[0, "groupid"]) 
  
    return policy
    
def getPolicyAllowlists(url, groupid):

    endpoint = url + '/v1/group/policies'
    payload = {
            "groupid" : f"{groupid}"
    }

    headers = {"X-APIKey": os.getenv('APIKEY')}
    payload = json.dumps(payload)

    response = requests.post(endpoint, headers=headers, data=payload, verify=False)
    result = json.loads(response.text)
    data = pd.DataFrame(result["response"]["applications"])
    return data

def getPolicyName(url, groupid):

    endpoint = url + '/v1/group/'
    payload = {
            
    }

    headers = {"X-APIKey": os.getenv('APIKEY')}
    payload = json.dumps(payload)

    response = requests.post(endpoint, headers=headers, data=payload, verify=False)
    result = json.loads(response.text)
    data = pd.DataFrame(result["response"]["groups"])
    name = data.loc[data['groupid'] == f"{groupid}", 'name'].values[0]
    return name


def devicehistory(url, outputjson: bool):
	endpoint = url + '/v1/getexechistory'
	print("\n")
	print(ct.colorText("1. Today", "yellow"))
	print(ct.colorText("2. Last 24 Hours", "yellow"))
	print(ct.colorText("3. Past 7 Days", "yellow"))
	print(ct.colorText("4. Past 30 Days", "yellow"))
	print(ct.colorText("5. Custom Date Range","yellow"))
	choice = input(ct.colorText("\nSelect Date Range: ", "white"))
	today = datetime.date.today()
	today = today.strftime("%Y-%m-%d")
	if choice == '1':
		date_selected = today
	elif choice == '2':
		date_selected = datetime.date.today() - datetime.timedelta(days=1)
		date_selected = date_selected.strftime('%Y-%m-%d')
	elif choice == '3':
		date_selected = datetime.date.today() - datetime.timedelta(days=7)
		date_selected = date_selected.strftime('%Y-%m-%d')
	elif choice == '4':
		date_selected = datetime.date.today() - datetime.timedelta(days=30)
		date_selected = date_selected.strftime('%Y-%m-%d')
	elif choice == "5":
		print(ct.colorText("Please Input Dates as YYYY-MM-DD", "cyan"))
		date_selected = input(ct.colorText("From: ", "white"))
		today = input(ct.colorText("Date To: ", "white"))
	print(ct.colorText("WARNING: Device Name is Case Sensitive", "red"))
	device = input(ct.colorText("Enter Device Name: ", "white"))
	payload_dict = {
		"datefrom": date_selected,
		"dateto": today,
		"hostname": device
	}
	payload = json.dumps(payload_dict)
	print(ct.colorText(payload, "green"))
	headers = {
		"X-APIKey": os.getenv('APIKEY')
	}

	response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)

	if outputjson:
		return response

	parse_text = json.loads(response.text)

	# Safely get exechistory
	exechistory = parse_text.get('response', {}).get('exechistory')

	if isinstance(exechistory, list):
		for block in exechistory:
			print(ct.colorText(f"Command: {block.get('commandline', 'N/A')}", "green"))
			print(ct.colorText(f"Date: {block.get('datetime', 'N/A')}", "green"))
			print(ct.colorText(f"Filename: {block.get('filename', 'N/A')}", "green"))
			print(ct.colorText(f"Policy Name: {block.get('policyname', 'N/A')}", "green"))
			print(ct.colorText(f"Hostname: {block.get('hostname', 'N/A')}", "green"))
			print(ct.colorText(f"Hash: {block.get('sha256', 'N/A')}", "green"))
			print("\n")
	else:
		print(ct.colorText("No execution history found or data is not in expected format.", "red"))