Zarithas
e1e0cb7ac7
- Implemented version checking system with update notifications - Integrated Git for fetching and downloading the latest version - Added statistics updates - Removed unused code across the project - Condensed project structure - Updated README - Cleaned up UI
79 lines
2.6 KiB
Python
79 lines
2.6 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 gc
|
|
import json
|
|
import logging
|
|
|
|
from bson import ObjectId
|
|
import pandas as pd
|
|
|
|
from services.API import AirlockAPIWrapper
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def getExecutions(api: AirlockAPIWrapper, policy, type, days):
|
|
import airlock_libs
|
|
|
|
executionhist_policy = pd.DataFrame()
|
|
exehist = airlock_libs.pull_policy_exec_histories(api, policy.name, str(type), days)
|
|
if exehist is not None:
|
|
data = json.loads(exehist)
|
|
executionhist_policy = pd.DataFrame(data["response"]["exechistories"])
|
|
if not executionhist_policy.empty:
|
|
executionhist_policy = executionhist_policy[
|
|
[
|
|
"datetime",
|
|
"sha256",
|
|
"publisher",
|
|
"filename",
|
|
"hostname",
|
|
"username",
|
|
"pprocess",
|
|
"gprocess",
|
|
"commandline",
|
|
]
|
|
]
|
|
executionhist_policy["policy"] = policy # Add policy column here
|
|
executionhist_policy = executionhist_policy.drop_duplicates(
|
|
subset=["sha256", "filename", "hostname"]
|
|
)
|
|
executionhist_policy = executionhist_policy.sort_values(
|
|
by=["sha256", "filename"]
|
|
)
|
|
logger.debug(f"Staging of Execution history for policy: {policy} is complete")
|
|
|
|
del data
|
|
del exehist
|
|
gc.collect()
|
|
return executionhist_policy
|
|
|
|
|
|
def skipback(days):
|
|
"""
|
|
Generate a MongoDB ObjectId for a given number of days ago from today.
|
|
"""
|
|
adjusted_days = days
|
|
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)
|