Driven-Element
76 lines
3.2 KiB
Python
76 lines
3.2 KiB
Python
# 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 time
|
|
import utils.colortext as ct
|
|
|
|
def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
|
|
|
|
headers = {
|
|
"X-APIKey": os.getenv('APIKEY')
|
|
}
|
|
checkpoint = '000000000000000000000000'
|
|
json_output = {'error': 'Success', 'response': {'exechistories': []}}
|
|
while True:
|
|
json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers)
|
|
if not json_response_data['response']['exechistories']:
|
|
break
|
|
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"Date Greater than 30 Days, Stepping to new Checkpoint. {item['checkpoint']}", "blue"))
|
|
else:
|
|
if (datetime.date.today() - datetime.timedelta(days=10) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
|
|
pass
|
|
else:
|
|
for output in json_response_data['response']['exechistories']:
|
|
json_output['response']['exechistories'].append(output)
|
|
json_output = json.dumps(json_output)
|
|
if outputjson == True:
|
|
return json_output
|
|
|
|
def checkpoint_stomper(checkpoint, url, policy, headers):
|
|
endpoint = url + '/v1/logging/exechistories'
|
|
payload_dict = {
|
|
"type":[1,2,6,7],
|
|
"checkpoint": checkpoint,
|
|
"policy": [policy]
|
|
}
|
|
payload = json.dumps(payload_dict)
|
|
response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
|
|
parse_text = json.loads(response.text)
|
|
return parse_text
|
|
|
|
def listPolicies(url):
|
|
endpoint = url + '/v1/group'
|
|
print(ct.colorText("[+] Grabbing All Policies", "magenta"))
|
|
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: ", "yellow"))
|
|
choice = int(choice) - 1
|
|
return choice, policiesnames |