Merge pull request #3 from SpecterOps/fixes_aug_22_23

Fixes aug 22 23
2026-06-08 12:36:42 +00:00 · 2023-08-25 18:54:29 -07:00
parent 83f2db824c 0a08cd9497
commit 65f5705e75
13 changed files with 445 additions and 283 deletions
@@ -33,6 +33,14 @@ Built on Kubernetes with scale in mind, our goal with Nemesis was to create a ce

 Nemesis aims to automate a number of repetitive tasks operators encounter on engagements, empower operators’ analytic capabilities and collective knowledge, and create structured and unstructured data stores of as much operational data as possible to help guide future research and facilitate offensive data analysis.

+## Nemesis Blog Posts:
+
+| Post Name                                   | Publication Date | Link                                                                               |
+|---------------------------------------------|------------------|------------------------------------------------------------------------------------|
+| *Hacking With Your Nemesis*                 | Aug 9, 2023      | https://posts.specterops.io/hacking-with-your-nemesis-7861f75fcab4                 |
+| *Challenges In Post-Exploitation Workflows* | Aug 2, 2023      | https://posts.specterops.io/challenges-in-post-exploitation-workflows-2b3469810fe9 |
+| *On (Structured) Data*                      | Jul 26, 2023     | https://posts.specterops.io/on-structured-data-707b7d9876c6                        |
+
 # Setup

 1. Ensure the hardware/software requisites are met and configuration values are completed [as described here in the setup](./docs/setup.md)
@@ -43,7 +51,7 @@ Nemesis aims to automate a number of repetitive tasks operators encounter on eng

 In the root directory of the repo, use skaffold to start everything:
 ```
-skaffold run
+skaffold run  --port-forward
 ```

 Run `skaffold delete` to remove running pods.
@@ -1,6 +1,6 @@
 BSD 3-Clause License

-Copyright (c) 2021-2022, SpecterOps
+Copyright (c) 2023, SpecterOps
 All rights reserved.

 Redistribution and use in source and binary forms, with or without
@@ -15,30 +15,35 @@ MYTHIC_USERNAME=mythic_admin
 MYTHIC_PASSWORD=SuperSecretPassword
 REDIS_HOSTNAME=redis
 REDIS_PORT=6379
-NEMESIS_HTTP_SERVER=http://127.0.0.1:8000
+NEMESIS_HTTP_SERVER=http://172.16.111.187:8000
 NEMESIS_CREDS=nemesis:password
 ELASTICSEARCH_USER=elastic
 ELASTICSEARCH_PASSWORD=password
+MAX_FILE_SIZE=100000000
+EXPIRATION_DAYS=100
 ```

+**Note**: The `NEMESIS_CREDS` are the `basic_auth_user` / `basic_auth_password` from the nemesis.config or set during the ./nemesis-cli.py setup. `MAX_FILE_SIZE` is in bytes, and `EXPIRATION_DAYS` is the number of days until data will be expunged from backend storage.
+
+**Make sure the `NEMESIS_HTTP_SERVER` and `MYTHIC_IP` variables do not reference localhost or 127.0.0.1! They need to be reachable from a Docker container.**
+
 Once the environment variables are setup, you can launch the service by using `docker-compose`:

 ``` bash
 sudo docker-compose up --build
 ```

-### Verify Successful Start-Up
-
-
 ## Troubleshooting

 Logs can be seen from the docker container via `sudo docker logs mythic_nemesis_sync` and follow them with `sudo docker logs --follow mythic_nemesis_sync`.

 Ensure the host where `mythic_nemesis_sync` is running has network access to the Nemesis and Mythic servers.

-`mythic_nemesis_sync` uses an internal Redis database to sync what events have already been sent to Nemesis, avoiding duplicates.
+`mythic_nemesis_sync` uses an internal Redis database to sync what events have already been sent to Nemesis, avoiding duplicates. If the `mythic_nemesis_sync` service goes down, it *should* be safe to stand it back up - duplicates should be available long as nothing has forcefully stopped/deleted Mythic's Redis container.

-If the `mythic_nemesis_sync` service goes down, it is safe to stand it back up and avoid duplicates as long as nothing has forcefully stopped Mythic's Redis container.
+## Reprocessing Data
+
+The container uses Redis to keep a persistent store of Mythic data that's been submitted to Nemesis. If you want to reprocess data, set `CLEAR_REDIS=True` in settings.env to clear the Redis database. There will be a 30 second pause on startup with a warning message indicating aborting the standup will avoid clearing the database.

 ## References

@@ -1,6 +1,6 @@
 aiohttp==3.8.1
 redis==3.5.3
-#mythic==0.0.37
-mythic==0.1.0rc2
+mythic==0.1.5
 requests==2.28.1
-elasticsearch==8.4.1
+elasticsearch==8.4.1
+gql==3.4.1
@@ -1,10 +1,12 @@
-MYTHIC_IP=10.10.1.100
+MYTHIC_IP=172.16.111.186
 MYTHIC_PORT=7443
 MYTHIC_USERNAME=mythic_admin
-MYTHIC_PASSWORD=SuperSecretPassword
+MYTHIC_PASSWORD=password
 REDIS_HOSTNAME=redis
 REDIS_PORT=6379
-NEMESIS_HTTP_SERVER=http://127.0.0.1:8000
+NEMESIS_HTTP_SERVER=http://172.16.111.187:8080
 NEMESIS_CREDS=nemesis:password
-ELASTICSEARCH_USER=elastic
-ELASTICSEARCH_PASSWORD=password
+ELASTICSEARCH_USER=nemesis
+ELASTICSEARCH_PASSWORD=password
+MAX_FILE_SIZE=100000000
+EXPIRATION_DAYS=100
@@ -10,15 +10,19 @@ from dataclasses import dataclass
 from datetime import datetime, timedelta

 import aiohttp
+import gql
 import redis
 import requests
-from elasticsearch import Elasticsearch
+from elasticsearch import AuthenticationException, Elasticsearch
 # Mythic Sync Libraries
 # 3rd Party Libraries
 from mythic import mythic, mythic_classes
 from requests.auth import HTTPBasicAuth

-logging.basicConfig(format="%(levelname)s:%(message)s")
+# logging.basicConfig(format="%(levelname)s:%(message)s")
+logging.basicConfig(
+    format='%(levelname)s %(asctime)s %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S')
 mythic_sync_log = logging.getLogger("mythic_sync_logger")
 mythic_sync_log.setLevel(logging.DEBUG)

@@ -53,6 +57,15 @@ if REDIS_PORT is None:
    mythic_sync_log.error("REDIS_PORT must be supplied!\n")
    sys.exit(1)

+# If we want to reprocess everything, signal we want to clear Redis after startup
+clear_redis = False
+try:
+    clear_redis_str = f"{os.environ.get('CLEAR_REDIS')}".lower()
+    if clear_redis_str.startswith("t"):
+        clear_redis = True
+except:
+    pass
+
 # Redis connector
 rconn = None

@@ -116,7 +129,10 @@ def nemesis_post_data(data):
        basic = HTTPBasicAuth(basic_auth_parts[0], basic_auth_parts[1])
        r = requests.post(f"{NEMESIS_URL}/data", auth=basic, json=data)
        if r.status_code != 200:
-            mythic_sync_log.error(f"[nemesis_post_data] Error posting to Nemesis URL {NEMESIS_URL}/data ({r.status_code}) : {r.json()}")
+            try:
+                mythic_sync_log.error(f"[nemesis_post_data] Error posting to Nemesis URL {NEMESIS_URL}/data ({r.status_code}) : {r.text}")
+            except:
+                mythic_sync_log.error(f"[nemesis_post_data] Error posting to Nemesis URL {NEMESIS_URL}/data ({r.status_code})")
            return None
        else:
            return r.json()
@@ -262,18 +278,12 @@ async def handle_file(mythic_instance: mythic_classes.Mythic) -> None:
        The Mythic instance to be used to query the Mythic database
    """

-    # # clear Redis, for testing
-    # for key in rconn.keys('*'):
-    #     rconn.delete(key)
-
    try:
        start_id = rconn.get("last_file_id")
    except:
        rconn.mset({"last_file_id": 0})
        start_id = 0

-    start_id = 39
-
    nemesis_file_subscription = """
    subscription NemesisFileSubscription {
        filemeta_stream(cursor: {initial_value: {id: %s}}, batch_size: 5, where: {is_download_from_agent: {_eq: true}, complete: {_eq: true}, is_screenshot: {_eq: false}}) {
@@ -300,7 +310,7 @@ async def handle_file(mythic_instance: mythic_classes.Mythic) -> None:
        start_id
    )

-    mythic_sync_log.info("Starting subscription for file data")
+    mythic_sync_log.info(f"Starting subscription for file data, start_id: {start_id}")
    async for data in mythic.subscribe_custom_query(mythic=mythic_instance, query=nemesis_file_subscription):
        try:
            file_meta = data["filemeta_stream"][0]
@@ -360,85 +370,87 @@ async def handle_file(mythic_instance: mythic_classes.Mythic) -> None:
                file_data["size"] = file_size
                file_data["object_id"] = nemesis_file_id

-                # post to the Nemesis data API (`data`` needs to be an array of dictionaries!)
+                # post to the Nemesis data API (`data` needs to be an array of dictionaries!)
                resp = nemesis_post_data({"metadata": metadata, "data": [file_data]})

-                message_id = resp["object_id"]
-                mythic_sync_log.info(f"Nemesis message_id for submitted file_data: {message_id}")
+                # if the post works we get JSON/non-none
+                if resp:
+                    message_id = resp["object_id"]
+                    mythic_sync_log.info(f"Nemesis message_id for submitted file_data: {message_id}")

-                # mark this file as processed in Redis now that it was submitted
-                rconn.mset({redis_key: 1})
+                    # mark this file as processed in Redis now that it was submitted
+                    rconn.mset({redis_key: 1})

-                # mark this Mythic ID as the last processed file ID
-                last_file_id = rconn.get("last_file_id")
-                if file_id > last_file_id:
-                    rconn.mset({"last_file_id": file_id})
+                    # mark this Mythic ID as the last processed file ID
+                    last_file_id = rconn.get("last_file_id")
+                    if file_id > last_file_id:
+                        rconn.mset({"last_file_id": file_id})

-                # get the metadata for the processed file from Elasticsearch
-                file_metadata = await get_processed_file_metadata(message_id)
+                    # get the metadata for the processed file from Elasticsearch
+                    file_metadata = await get_processed_file_metadata(message_id)

-                if not file_metadata:
-                    mythic_sync_log.error("Couldn't retrieve metadata for processed file!")
-                    continue
+                    if not file_metadata:
+                        mythic_sync_log.error("Couldn't retrieve metadata for processed file!")
+                        continue

-                mythic_sync_log.debug(f"File metadata for {message_id} retrieved from Elastic")
+                    mythic_sync_log.debug(f"File metadata for {message_id} retrieved from Elastic")

-                # this is any metadata that we're doing to display as JSON for the "file_metadata" tag
-                file_metadata_display = {}
-                if "size" in file_metadata and file_metadata["size"]:
-                    file_metadata_display["size"] = file_metadata["size"]
-                if "isBinary" in file_metadata and file_metadata["isBinary"]:
-                    file_metadata_display["is_binary"] = "true"
-                if "isOfficeDoc" in file_metadata and file_metadata["isOfficeDoc"]:
-                    file_metadata_display["is_office_doc"] = "true"
-                if "magicType" in file_metadata and file_metadata["magicType"]:
-                    file_metadata_display["magic_type"] = file_metadata["magicType"]
-                if "nemesisFileType" in file_metadata and file_metadata["nemesisFileType"]:
-                    file_metadata_display["nemesis_file_type"] = file_metadata["nemesisFileType"]
+                    # this is any metadata that we're doing to display as JSON for the "file_metadata" tag
+                    file_metadata_display = {}
+                    if "size" in file_metadata and file_metadata["size"]:
+                        file_metadata_display["size"] = file_metadata["size"]
+                    if "isBinary" in file_metadata and file_metadata["isBinary"]:
+                        file_metadata_display["is_binary"] = "true"
+                    if "isOfficeDoc" in file_metadata and file_metadata["isOfficeDoc"]:
+                        file_metadata_display["is_office_doc"] = "true"
+                    if "magicType" in file_metadata and file_metadata["magicType"]:
+                        file_metadata_display["magic_type"] = file_metadata["magicType"]
+                    if "nemesisFileType" in file_metadata and file_metadata["nemesisFileType"]:
+                        file_metadata_display["nemesis_file_type"] = file_metadata["nemesisFileType"]

-                if "objectIdURL" in file_metadata and file_metadata["objectIdURL"]:
-                    file_metadata_display["Download File"] = file_metadata["objectIdURL"]
-                if "extractedSourceURL" in file_metadata and file_metadata["extractedSourceURL"]:
-                    file_metadata_display["Download Decompiled Source"] = file_metadata["extractedSourceURL"]
-                if "convertedPdfURL" in file_metadata and file_metadata["convertedPdfURL"]:
-                    file_metadata_display["View Converted PDF"] = file_metadata["convertedPdfURL"]
-                if "extractedPlaintextURL" in file_metadata and file_metadata["extractedPlaintextURL"]:
-                    file_metadata_display["Extracted Plaintext (Elastic)"] = file_metadata["extractedPlaintextURL"]
+                    if "objectIdURL" in file_metadata and file_metadata["objectIdURL"]:
+                        file_metadata_display["Download File"] = file_metadata["objectIdURL"]
+                    if "extractedSourceURL" in file_metadata and file_metadata["extractedSourceURL"]:
+                        file_metadata_display["Download Decompiled Source"] = file_metadata["extractedSourceURL"]
+                    if "convertedPdfURL" in file_metadata and file_metadata["convertedPdfURL"]:
+                        file_metadata_display["View Converted PDF"] = file_metadata["convertedPdfURL"]
+                    if "extractedPlaintextURL" in file_metadata and file_metadata["extractedPlaintextURL"]:
+                        file_metadata_display["Extracted Plaintext (Elastic)"] = file_metadata["extractedPlaintextURL"]

-                # TODO: fixate the index ID
-                kibana_file_link = f"{KIBANA_URL}/app/discover#/?_a=(filters:!((query:(match_phrase:(objectId:'{nemesis_file_id}')))))&_g=(time:(from:now-1y%2Fd,to:now))"
+                    # TODO: fixate the index ID
+                    kibana_file_link = f"{KIBANA_URL}/app/discover#/?_a=(filters:!((query:(match_phrase:(objectId:'{nemesis_file_id}')))))&_g=(time:(from:now-1y%2Fd,to:now))"

-                # update the comment for the file in Mythic to indicate it was processed
-                await mythic.update_file_comment(mythic_instance, file_uuid=mythic_file_id, comment="Processed by Nemesis 😈")
+                    # update the comment for the file in Mythic to indicate it was processed
+                    await mythic.update_file_comment(mythic_instance, file_uuid=mythic_file_id, comment="Processed by Nemesis 😈")

-                # add a basic metadata Mythic tag
-                await add_mythic_tag(mythic_instance, "file_metadata", filemeta_id=file_id, task_id=task_id, url=kibana_file_link, data=json.dumps(file_metadata_display))
+                    # add a basic metadata Mythic tag
+                    await add_mythic_tag(mythic_instance, "file_metadata", filemeta_id=file_id, task_id=task_id, url=kibana_file_link, data=json.dumps(file_metadata_display))

-                # custom tags
-                if ("containsDpapi" in file_metadata) and (file_metadata["containsDpapi"]):
-                    await add_mythic_tag(mythic_instance, "contains_dpapi", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)
+                    # custom tags
+                    if ("containsDpapi" in file_metadata) and (file_metadata["containsDpapi"]):
+                        await add_mythic_tag(mythic_instance, "contains_dpapi", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)

-                if ("parsedData" in file_metadata) and ("hasParsedCredentials" in file_metadata["parsedData"]) and file_metadata["parsedData"]["hasParsedCredentials"]:
-                    await add_mythic_tag(mythic_instance, "parsed_credentials", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)
+                    if ("parsedData" in file_metadata) and ("hasParsedCredentials" in file_metadata["parsedData"]) and file_metadata["parsedData"]["hasParsedCredentials"]:
+                        await add_mythic_tag(mythic_instance, "parsed_credentials", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)

-                if ("analysis" in file_metadata) and ("dotnetDeserialization" in file_metadata["analysis"]) and (file_metadata["analysis"]["dotnetDeserialization"]["hasDeserialization"] == 1):
-                    await add_mythic_tag(mythic_instance, "deserialization", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)
+                    if ("analysis" in file_metadata) and ("dotnetDeserialization" in file_metadata["analysis"]) and (file_metadata["analysis"]["dotnetDeserialization"]["hasDeserialization"] == 1):
+                        await add_mythic_tag(mythic_instance, "deserialization", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)

-                if ("parsedData" in file_metadata) and ("isEncrypted" in file_metadata["parsedData"]) and file_metadata["parsedData"]["isEncrypted"]:
-                    await add_mythic_tag(mythic_instance, "encrypted", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)
+                    if ("parsedData" in file_metadata) and ("isEncrypted" in file_metadata["parsedData"]) and file_metadata["parsedData"]["isEncrypted"]:
+                        await add_mythic_tag(mythic_instance, "encrypted", filemeta_id=file_id, task_id=task_id, url=kibana_file_link)

-                if ("yaraMatches" in file_metadata) and (file_metadata["yaraMatches"]):
-                    rule_names = ", ".join(file_metadata["yaraMatches"])
-                    data = json.dumps({"rule_names": rule_names})
-                    await add_mythic_tag(mythic_instance, "yara_matches", filemeta_id=file_id, task_id=task_id, url=kibana_file_link, data=data)
+                    if ("yaraMatches" in file_metadata) and (file_metadata["yaraMatches"]):
+                        rule_names = ", ".join(file_metadata["yaraMatches"])
+                        data = json.dumps({"rule_names": rule_names})
+                        await add_mythic_tag(mythic_instance, "yara_matches", filemeta_id=file_id, task_id=task_id, url=kibana_file_link, data=data)

-                if ("noseyparker" in file_metadata) and (file_metadata["noseyparker"]):
-                    rule_names_dict = {}
-                    for match in file_metadata["noseyparker"]["ruleMatches"]:
-                        rule_names_dict[match["ruleName"]] = True
-                    rule_names = ", ".join(rule_names_dict.keys())
-                    data = json.dumps({"rule_names": rule_names})
-                    await add_mythic_tag(mythic_instance, "noseyparker", filemeta_id=file_id, task_id=task_id, url=kibana_file_link, data=data)
+                    if ("noseyparker" in file_metadata) and (file_metadata["noseyparker"]):
+                        rule_names_dict = {}
+                        for match in file_metadata["noseyparker"]["ruleMatches"]:
+                            rule_names_dict[match["ruleName"]] = True
+                        rule_names = ", ".join(rule_names_dict.keys())
+                        data = json.dumps({"rule_names": rule_names})
+                        await add_mythic_tag(mythic_instance, "noseyparker", filemeta_id=file_id, task_id=task_id, url=kibana_file_link, data=data)

        except Exception:
            mythic_sync_log.exception(
@@ -474,6 +486,15 @@ async def handle_filebrowser(mythic_instance: mythic_classes.Mythic) -> None:
            parent_path_text
            timestamp
            can_have_children
+            task {
+                callback {
+                    agent_callback_id
+                    operation {
+                        name
+                    }
+                }
+                id
+            }
            metadata
        }
    }
@@ -481,7 +502,7 @@ async def handle_filebrowser(mythic_instance: mythic_classes.Mythic) -> None:
        start_id
    )

-    mythic_sync_log.info("Starting subscription for file browser information")
+    mythic_sync_log.info(f"Starting subscription for file browser information, start_id: {start_id}")
    async for data in mythic.subscribe_custom_query(mythic=mythic_instance, query=nemesis_filebrowser_subscription):

        # group by the agent ID
@@ -492,12 +513,12 @@ async def handle_filebrowser(mythic_instance: mythic_classes.Mythic) -> None:
        for file in files:
            mythic_id = file["id"]
            redis_key = f"filebrowser{mythic_id}"
-
            try:
                redis_entry_id = rconn.get(redis_key)
            except:
                redis_entry_id = None

+            # if this key is _not_ already processed
            if not redis_entry_id:
                callback_id = file["task"]["callback"]["agent_callback_id"]

@@ -507,7 +528,7 @@ async def handle_filebrowser(mythic_instance: mythic_classes.Mythic) -> None:
                    metadata["agent_id"] = callback_id
                    metadata["agent_type"] = "mythic"
                    metadata["automated"] = True
-                    metadata["data_type"] = "process"
+                    metadata["data_type"] = "file_information"
                    metadata["expiration"] = convert_timestamp(file["timestamp"], EXPIRATION_DAYS)
                    metadata["source"] = file["host"]
                    metadata["project"] = file["task"]["callback"]["operation"]["name"]
@@ -519,24 +540,46 @@ async def handle_filebrowser(mythic_instance: mythic_classes.Mythic) -> None:

                file_data = {}
                file_data["path"] = file["full_path_text"].replace("\\", "/")
-                file_data["size"] = file["metadata"]["size"]

-                if file["can_have_children"]:
+                if "metadata" in file and "size" in file["metadata"]:
+                    file_data["size"] = file["metadata"]["size"]
+
+                if "can_have_children" in file and file["can_have_children"]:
                    file_data["type"] = "folder"
                else:
                    file_data["type"] = "file"

-                if "access_time" in file["metadata"] and file["metadata"]["access_time"]:
-                    file_data["access_time"] = convert_timestamp(file["metadata"]["access_time"])
-                if "modify_time" in file["metadata"] and file["metadata"]["modify_time"]:
-                    file_data["modification_time"] = convert_timestamp(file["metadata"]["modify_time"])
+                if "metadata" in file and "access_time" in file["metadata"] and file["metadata"]["access_time"]:
+                    access_time = file["metadata"]["access_time"]
+                    if isinstance(access_time, int):
+                        file_data["access_time"] = datetime.fromtimestamp(access_time // 1000).strftime("%Y-%m-%dT%H:%M:%S.000Z")
+                    else:
+                        file_data["access_time"] = convert_timestamp(access_time)
+                if "metadata" in file and "modify_time" in file["metadata"] and file["metadata"]["modify_time"]:
+                    modify_time = file["metadata"]["modify_time"]
+                    if isinstance(modify_time, int):
+                        file_data["modification_time"] = datetime.fromtimestamp(modify_time // 1000).strftime("%Y-%m-%dT%H:%M:%S.000Z")
+                    else:
+                        file_data["modification_time"] = convert_timestamp(modify_time)

-                # TODO: translate file["metadata"]["permissions"] to sddl
+                # TODO: translate file["metadata"]["permissions"] to sddl for Windows machines, if possible
+
+                # handle *nix permissions
+                if "metadata" in file and "permissions" in file["metadata"] and len(file["metadata"]["permissions"]) > 0 \
+                    and "permissions" in file["metadata"]["permissions"][0] and file["metadata"]["permissions"][0]["permissions"]:
+                    try:
+                        persmission_json = json.loads(file["metadata"]["permissions"][0]["permissions"])
+                        if "user" in persmission_json:
+                            owner = persmission_json["user"]
+                            file_data["owner"] = owner
+                    except Exception as e:
+                        pass

                all_data[callback_id]["data"].append(file_data)

                # mark this file browser entry as seen
                # TODO: does this need to be after the nemesis_post_data call?
+                #       but when how do we handle last_filebrowser_id...
                rconn.mset({redis_key: 1})

                # mark this Mythic ID as the last processed process ID
@@ -547,7 +590,10 @@ async def handle_filebrowser(mythic_instance: mythic_classes.Mythic) -> None:
        # for each unique agent ID, issue one request with all batched file browser information
        #   but using the same metadata entry
        for key in all_data:
-            nemesis_post_data(all_data[key])
+            resp = nemesis_post_data(all_data[key])
+            if resp:
+                message_id = resp["object_id"]
+                mythic_sync_log.info(f"Nemesis message_id for submitted file listing data: {message_id}")


 async def handle_process(mythic_instance: mythic_classes.Mythic, chunk_size: int = 100) -> None:
@@ -561,10 +607,6 @@ async def handle_process(mythic_instance: mythic_classes.Mythic, chunk_size: int
        The number of process results to handle at a time.
    """

-    # # clear Redis, for testing
-    # for key in rconn.keys("*"):
-    #     rconn.delete(key)
-
    try:
        start_id = rconn.get("last_process_id")
    except:
@@ -590,7 +632,7 @@ async def handle_process(mythic_instance: mythic_classes.Mythic, chunk_size: int
    }
    """ % (chunk_size, start_id)

-    mythic_sync_log.info("Starting subscription for process data")
+    mythic_sync_log.info(f"Starting subscription for process data, start_id: {start_id}")
    async for data in mythic.subscribe_custom_query(mythic=mythic_instance, query=nemesis_process_subscription):

        # group by the callback ID
@@ -666,6 +708,7 @@ async def handle_process(mythic_instance: mythic_classes.Mythic, chunk_size: int

                # mark this process entry as seen
                # TODO: does this need to be after the nemesis_post_data call?
+                #       but when how do we handle last_filebrowser_id...
                rconn.mset({redis_key: 1})

                # mark this Mythic ID as the last processed process ID
@@ -677,44 +720,45 @@ async def handle_process(mythic_instance: mythic_classes.Mythic, chunk_size: int
        #   but using the same metadata entry
        for key in all_data:
            resp = nemesis_post_data(all_data[key])
-            message_id = resp["object_id"]
-            mythic_sync_log.info(f"Nemesis message_id for submitted process data: {message_id}")
+            if resp:
+                message_id = resp["object_id"]
+                mythic_sync_log.info(f"Nemesis message_id for submitted process data: {message_id}")

-            # get the metadata for the processed file from Elasticsearch
-            nemesis_processes = await get_process_metadata(message_id, chunk_size)
+                # get the metadata for the processed file from Elasticsearch
+                nemesis_processes = await get_process_metadata(message_id, chunk_size)

-            for nemesis_process in nemesis_processes:
-                if "name" in nemesis_process["origin"]:
-                    name = nemesis_process["origin"]["name"]
-                else:
-                    name = ""
+                for nemesis_process in nemesis_processes:
+                    if "name" in nemesis_process["origin"]:
+                        name = nemesis_process["origin"]["name"]
+                    else:
+                        name = ""

-                if "processId" in nemesis_process["origin"]:
-                    process_id = nemesis_process["origin"]["processId"]
-                else:
-                    process_id = ""
+                    if "processId" in nemesis_process["origin"]:
+                        process_id = nemesis_process["origin"]["processId"]
+                    else:
+                        process_id = ""

-                tag_calls = []
-                key = f"{name}{process_id}"
-                if key in mythic_process_lookup_table[callback_id]:
-                    category = nemesis_process["category"]["category"]
+                    tag_calls = []
+                    key = f"{name}{process_id}"
+                    if key in mythic_process_lookup_table[callback_id]:
+                        category = nemesis_process["category"]["category"]

-                    if category != "Unknown":
-                        mythictree_id = mythic_process_lookup_table[callback_id][key]
+                        if category != "Unknown":
+                            mythictree_id = mythic_process_lookup_table[callback_id][key]

-                        if "description" in nemesis_process["category"]:
-                            description = nemesis_process["category"]["description"]
-                            data = json.dumps({"description": description})
-                        else:
-                            data = ""
+                            if "description" in nemesis_process["category"]:
+                                description = nemesis_process["category"]["description"]
+                                data = json.dumps({"description": description})
+                            else:
+                                data = ""

-                        tag_calls.append(add_mythic_tag(mythic_instance, category, mythictree_id=mythictree_id, data=data))
+                            tag_calls.append(add_mythic_tag(mythic_instance, category, mythictree_id=mythictree_id, data=data))

-                await asyncio.gather(*tag_calls)
+                    await asyncio.gather(*tag_calls)


 async def add_mythic_tag(mythic_instance: mythic_classes.Mythic, tag_name: str, source: str = "Nemesis", filemeta_id: int = -1, mythictree_id: int = -1, task_id: int = -1, url: str = "", data: str = ""):
-    """Adds a file or process tag from an existing tag type."""
+    """Adds a file or process tag to Mythic from an existing created tag type."""

    if filemeta_id != -1:
        filemeta_ids = [filemeta_id]
@@ -752,36 +796,55 @@ async def create_tag_types(mythic_instance: mythic_classes.Mythic):

 async def wait_for_service() -> None:
    """Wait for an HTTP session to be established with Mythic."""
-    while True:
-        mythic_sync_log.info(f"Attempting to connect to {MYTHIC_URL}")
-        async with aiohttp.ClientSession() as session:
-            async with session.get(MYTHIC_URL, ssl=False) as resp:
-                if resp.status != 200:
-                    mythic_sync_log.warning(
-                        "Expected 200 OK and received HTTP code %s while trying to connect to Mythic, trying again in %s seconds...",
-                        resp.status,
-                        WAIT_TIMEOUT,
-                    )
-                    await asyncio.sleep(WAIT_TIMEOUT)
-                    continue
-        return
+    retries = 5
+    while retries >= 0:
+        retries -= 1
+        if retries == 0:
+            mythic_sync_log.error("Out of retries for connecting to Mythic")
+            return False
+        mythic_sync_log.info(f"Attempting to connect to Mythic URL: {MYTHIC_URL}")
+        try:
+            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=WAIT_TIMEOUT)) as session:
+                async with session.get(MYTHIC_URL, ssl=False) as resp:
+                    if resp.status != 200:
+                        mythic_sync_log.warning(
+                            "Expected 200 OK and received HTTP code %s while trying to connect to Mythic, trying again in %s seconds...",
+                            resp.status,
+                            WAIT_TIMEOUT,
+                        )
+                        await asyncio.sleep(WAIT_TIMEOUT)
+                        continue
+            return True
+        except asyncio.exceptions.TimeoutError as e:
+            mythic_sync_log.warning(f"Timeout error connecting to Mythic URL: '{MYTHIC_URL}'. Trying again in {WAIT_TIMEOUT} seconds, retries: {retries}.")


 async def wait_for_redis() -> None:
    """Wait for a connection to be established with Mythic's Redis container."""
    global rconn
-    while True:
+    retries = 5
+    while retries >= 0:
+        retries -= 1
+        if retries == 0:
+            mythic_sync_log.exception("Out of retries for connecting to Redis")
+            return False
        try:
            rconn = redis.Redis(host=REDIS_HOSTNAME, port=REDIS_PORT, db=1)
            # we're only using ints for our redis DB
            rconn.set_response_callback("GET", int)
-            return
+            if clear_redis:
+                mythic_sync_log.warning("CLEAR_REDIS env variable set, clearing Redis database in 30 seconds (stop standup to cancel)...")
+                await asyncio.sleep(30)
+                for key in rconn.keys('*'):
+                    rconn.delete(key)
+            return True
        except Exception:
            mythic_sync_log.exception(
-                "Encountered an exception while trying to connect to Redis, %s:%s, trying again in %s seconds...",
+                "Encountered an exception while trying to connect to Redis, %s:%s, trying again in %s seconds, retries: %s...",
                REDIS_HOSTNAME,
                REDIS_PORT,
                WAIT_TIMEOUT,
+                retries
            )
            await asyncio.sleep(WAIT_TIMEOUT)
            continue
@@ -789,7 +852,13 @@ async def wait_for_redis() -> None:

 async def wait_for_authentication() -> mythic_classes.Mythic:
    """Wait for authentication with Mythic to complete."""
-    while True:
+    retries = 5
+    while retries >= 0:
+        retries -= 1
+        if retries == 0:
+            mythic_sync_log.exception("Out of retries for authenticating to Mythic")
+            return False
+
        # If ``MYTHIC_API_KEY`` is not set in the environment, then authenticate with user credentials
        if len(MYTHIC_API_KEY) == 0:
            mythic_sync_log.info(
@@ -804,22 +873,18 @@ async def wait_for_authentication() -> mythic_classes.Mythic:
                    server_ip=MYTHIC_IP,
                    server_port=MYTHIC_PORT,
                    ssl=True,
-                    timeout=-1,
+                    logging_level=logging.ERROR,
+                    timeout=-1
                )
-            except Exception:
-                mythic_sync_log.exception(
-                    "Encountered an exception while trying to authenticate to Mythic, trying again in %s seconds...",
-                    WAIT_TIMEOUT,
+            except gql.transport.exceptions.TransportQueryError as e:
+                mythic_sync_log.error(
+                    f"Encountered an exception while trying to authenticate to Mythic, trying again in {WAIT_TIMEOUT} seconds: '{e.errors[0]['message']}'"
                )
                await asyncio.sleep(WAIT_TIMEOUT)
                continue
-            try:
-                # await mythic.get_me(mythic=mythic_instance) # TODO: replace?
-                pass
-            except Exception:
-                mythic_sync_log.exception(
-                    "Encountered an exception while trying to authenticate to Mythic, trying again in %s seconds...",
-                    WAIT_TIMEOUT,
+            except Exception as e:
+                mythic_sync_log.error(
+                    f"Encountered an exception while trying to authenticate to Mythic, trying again in {WAIT_TIMEOUT} seconds: {e}"
                )
                await asyncio.sleep(WAIT_TIMEOUT)
                continue
@@ -844,13 +909,18 @@ async def wait_for_authentication() -> mythic_classes.Mythic:
                    server_ip=MYTHIC_IP,
                    server_port=MYTHIC_PORT,
                    ssl=True,
-                    global_timeout=-1,
+                    logging_level=logging.ERROR,
+                    timeout=-1
                )
-                # await mythic.get_me(mythic=mythic_instance) # TODO: replace?
-            except Exception:
-                mythic_sync_log.exception(
-                    "Failed to authenticate with the Mythic API token, trying again in %s seconds...",
-                    WAIT_TIMEOUT,
+            except gql.transport.exceptions.TransportQueryError as e:
+                mythic_sync_log.error(
+                    f"Encountered an exception while trying to authenticate to Mythic with an API token, trying again in {WAIT_TIMEOUT} seconds: '{e.errors[0]['message']}'"
+                )
+                await asyncio.sleep(WAIT_TIMEOUT)
+                continue
+            except Exception as e:
+                mythic_sync_log.error(
+                    f"Encountered an exception while trying to authenticate to Mythic with an API token, trying again in {WAIT_TIMEOUT} seconds: {e}"
                )
                await asyncio.sleep(WAIT_TIMEOUT)
                continue
@@ -861,32 +931,77 @@ async def wait_for_authentication() -> mythic_classes.Mythic:
 async def wait_for_elasticsearch() -> None:
    """Wait for a connection to be established with Nemesis' Elasticsearch container."""
    global es_client
-    while True:
+
+    retries = 5
+    while retries >= 0:
+        retries -= 1
+        if retries == 0:
+            mythic_sync_log.error("Out of retries for reaching the Elasticsearch endpoint")
+            return False
        try:
-            es_client = Elasticsearch(ELASTICSEARCH_URL, basic_auth=(ELASTICSEARCH_USER, ELASTICSEARCH_PASSWORD), verify_certs=False)
-            es_client.info()
-            return
-        except Exception:
-            mythic_sync_log.exception(
-                "Encountered an exception while trying to connect to Elasticsearch %s, trying again in %s seconds...",
-                ELASTICSEARCH_URL,
-                WAIT_TIMEOUT,
-            )
+            get = requests.get(ELASTICSEARCH_URL, auth=(ELASTICSEARCH_USER, ELASTICSEARCH_PASSWORD))
+            status_code = get.status_code
+            if status_code == 200:
+                mythic_sync_log.info("Successfully reached the Elasticsearch endpoint")
+                break
+            elif status_code == 401:
+                mythic_sync_log.warning(f"Error reaching the Elasticsearch endpoint, code {status_code}, likely incorrect ELASTICSEARCH_USER / ELASTICSEARCH_PASSWORD")
+                return False
+            else:
+                mythic_sync_log.warning(
+                    f"Failed to reach the Elasticsearch endpoint {ELASTICSEARCH_URL}, status: {status_code}, retries: {retries}. Trying again in {WAIT_TIMEOUT} seconds"
+                    )
+        except requests.exceptions.RequestException as e:
+            mythic_sync_log.warning(
+                f"Exception reaching the Elasticsearch endpoint {ELASTICSEARCH_URL}. Trying again in {WAIT_TIMEOUT} seconds, retries: {retries}. Exception: {e}."
+                )
            await asyncio.sleep(WAIT_TIMEOUT)
            continue

+    try:
+        es_client = Elasticsearch(ELASTICSEARCH_URL, basic_auth=(ELASTICSEARCH_USER, ELASTICSEARCH_PASSWORD), verify_certs=False)
+        es_client.info()
+        mythic_sync_log.info("Successfully authenticated to the Elasticsearch endpoint")
+    except AuthenticationException as e:
+        mythic_sync_log.warning(f"Error authenticating to Elasticsearch, code {e.status_code}, likely incorrect ELASTICSEARCH_USER / ELASTICSEARCH_PASSWORD")
+        return False
+    except Exception as e:
+        mythic_sync_log.warning(
+            f"Exception authenticating to Elasticsearch endpoint {ELASTICSEARCH_URL}. Trying again in {WAIT_TIMEOUT} seconds, retries: {retries}. Exception: {e}"
+            )
+        return False
+
+    return True
+

 async def scripting():
    while True:
-        await wait_for_redis()
-        mythic_sync_log.info("Successfully connected to Redis")
-        await wait_for_elasticsearch()
-        mythic_sync_log.info("Successfully connected to Nemesis-Elasticsearch")
-        await wait_for_service()
-        mythic_sync_log.info(f"Successfully connected to {MYTHIC_URL}")
-        mythic_sync_log.info("Trying to authenticate to Mythic")
+
+        if await wait_for_redis():
+            mythic_sync_log.info("Successfully connected to Redis")
+        else:
+            await asyncio.sleep(WAIT_TIMEOUT)
+            continue
+
+        if await wait_for_elasticsearch():
+            mythic_sync_log.info("Successfully connected to Nemesis-Elasticsearch")
+        else:
+            await asyncio.sleep(WAIT_TIMEOUT)
+            continue
+
+        if await wait_for_service():
+            mythic_sync_log.info(f"Successfully connected to Mythic URL {MYTHIC_URL}")
+        else:
+            await asyncio.sleep(WAIT_TIMEOUT)
+            continue
+
        mythic_instance = await wait_for_authentication()
-        mythic_sync_log.info("Successfully authenticated to Mythic")
+        if mythic_instance:
+            mythic_sync_log.info("Successfully authenticated to Mythic")
+        else:
+            await asyncio.sleep(WAIT_TIMEOUT)
+            continue
+
        # create our initial tags
        mythic_sync_log.info("Creating tag types")
        await create_tag_types(mythic_instance)
@@ -895,11 +1010,12 @@ async def scripting():
        try:
            _ = await asyncio.gather(
                handle_file(mythic_instance=mythic_instance),
-                # handle_process(mythic_instance=mythic_instance),
-                # handle_filebrowser(mythic_instance=mythic_instance),
+                handle_process(mythic_instance=mythic_instance),
+                handle_filebrowser(mythic_instance=mythic_instance),
            )
        except Exception:
            mythic_sync_log.exception("Encountered an exception while subscribing to tasks and responses, restarting...")

+        await asyncio.sleep(WAIT_TIMEOUT)

 asyncio.run(scripting())
@@ -14,8 +14,8 @@ from enrichment.lib.nemesis_db import NemesisDb
 from enrichment.services.text_extractor import TikaTextExtractor
 from enrichment.settings import EnrichmentSettings
 from enrichment.tasks.chromium_cookie import ChromiumCookie
-from enrichment.tasks.dpapi.dpapi import Dpapi
 from enrichment.tasks.data_expunge import DataExpunge
+from enrichment.tasks.dpapi.dpapi import Dpapi
 from enrichment.tasks.elastic_connector import ElasticConnector
 from enrichment.tasks.file_processor import FileProcessor
 from enrichment.tasks.postgres_connector.postgres_connector import (
@@ -534,6 +534,7 @@ class Container(containers.DeclarativeContainer):
        FileProcessor,
        alerter_service,
        storage_service,
+        database,
        text_extractor,
        # URIs
        config.crack_list_uri,
@@ -512,6 +512,10 @@ class NemesisDbInterface(ABC):
    async def update_decrypted_chromium_cookie(self, unique_db_id: UUID4, value_dec: str) -> None:
        pass

+    @abstractmethod
+    async def is_file_processed(self, file_sha256: str) -> bool:
+        pass
+

 NemesisDbT = TypeVar("NemesisDbT", bound="NemesisDb")
 _Record = TypeVar("_Record", bound=asyncpg.protocol.Record)
@@ -1486,6 +1490,15 @@ class NemesisDb(NemesisDbInterface):
        async with self.pool.acquire() as conn:
            await conn.execute(query, value_dec, unique_db_id)

+    async def is_file_processed(self, file_sha256: str) -> bool:
+        """Takes the sha256 of a file and returns whether the file has already been processed."""
+
+        async with self.pool.acquire() as conn:
+            return await conn.fetch(
+                "SELECT EXISTS (SELECT true FROM nemesis.file_data_enriched WHERE sha256 = $1)",
+                file_sha256
+            )
+
    # async def sanitize_identifier(self, identifier: str) -> str:
    #     """Sanitizes a postgres column names to make it safe for use in dynamic queries

@@ -20,6 +20,7 @@ import nemesispb.nemesis_pb2 as pb
 import structlog
 import yara
 from binaryornot.check import is_binary
+from enrichment.lib.nemesis_db import NemesisDb
 from enrichment.services.text_extractor import TextExtractorInterface
 from nemesiscommon.messaging import (MessageQueueConsumerInterface,
                                     MessageQueueProducerInterface)
@@ -39,6 +40,7 @@ logger = structlog.get_logger(module=__name__)
 class FileProcessor(TaskInterface):
    alerter: AlerterInterface
    storage: StorageInterface
+    db: NemesisDb
    text_extractor: TextExtractorInterface

    # URIs
@@ -75,6 +77,7 @@ class FileProcessor(TaskInterface):
        self,
        alerter: AlerterInterface,
        storage: StorageInterface,
+        db: NemesisDb,
        text_extractor: TextExtractorInterface,
        # URIs
        crack_list_uri: str,
@@ -107,6 +110,7 @@ class FileProcessor(TaskInterface):
    ):
        self.alerter = alerter
        self.storage = storage
+        self.db = db
        self.text_extractor = text_extractor

        self.crack_list_uri = crack_list_uri
@@ -272,6 +276,18 @@ class FileProcessor(TaskInterface):
        file_data.is_source_code = is_source_code
        file_data.nemesis_file_type = "unknown"

+        try:
+            file_previously_processed = (await self.db.is_file_processed(file_data.hashes.sha256))[0][0]
+        except:
+            file_previously_processed = False
+
+        if file_previously_processed:
+            await logger.ainfo(
+                "File has already been processed.",
+                file_name=file_data.name,
+                sha256=file_data.hashes.sha256,
+            )
+
        ###########################################################
        #
        # Nemesis-defined file format parsing
@@ -371,13 +387,13 @@ class FileProcessor(TaskInterface):
                    file_data.parsed_data.CopyFrom(parsed)
                    try:
                        if file_data.parsed_data.has_parsed_credentials:
-                            await self.alerter.file_data_alert(
-                                file_data=file_data,
-                                metadata=metadata,
-                                title="Parsed Credentials",
-                                text="File is a known type and has some form of parsed credentials!",
-                            )
-
+                            if not file_previously_processed:
+                                await self.alerter.file_data_alert(
+                                    file_data=file_data,
+                                    metadata=metadata,
+                                    title="Parsed Credentials",
+                                    text="File is a known type and has some form of parsed credentials!",
+                                )
                    except:
                        pass

@@ -470,11 +486,12 @@ class FileProcessor(TaskInterface):

                    if noseyparker_output and len(noseyparker_output.rule_matches) > 0:
                        file_data.noseyparker.CopyFrom(noseyparker_output)
-                        await self.alerter.file_data_alert(
-                            file_data=file_data,
-                            title="NoseyParker Results",
-                            metadata=metadata,
-                        )
+                        if not file_previously_processed:
+                            await self.alerter.file_data_alert(
+                                file_data=file_data,
+                                title="NoseyParker Results",
+                                metadata=metadata,
+                            )
                except Exception as e:
                    await logger.aexception(e, message="Noseyparker scanning failed")
                    enrichments_failure.append(constants.E_NOSEYPARKER_SCAN)
@@ -492,11 +509,12 @@ class FileProcessor(TaskInterface):

                file_data.contains_dpapi = True

-                await self.alerter.file_data_alert(
-                    file_data=file_data,
-                    metadata=metadata,
-                    title="DPAPI data present",
-                )
+                if not file_previously_processed:
+                    await self.alerter.file_data_alert(
+                        file_data=file_data,
+                        metadata=metadata,
+                        title="DPAPI data present",
+                    )

                # carve any DPAPI blobs
                carved = await helpers.carve_dpapi_blobs_from_file(file_path_on_disk, file_uuid_str, metadata)
@@ -590,12 +608,13 @@ class FileProcessor(TaskInterface):
                urls = urls.replace(".", "[.]")
                text += f"*Rule {rule} :* {urls}\n"

-            await self.alerter.file_data_alert(
-                file_data=file_data,
-                metadata=metadata,
-                title="Possible canaries detected",
-                text=text,
-            )
+            if not file_previously_processed:
+                await self.alerter.file_data_alert(
+                    file_data=file_data,
+                    metadata=metadata,
+                    title="Possible canaries detected",
+                    text=text,
+                )

        # run Yara on the file
        yara_results = await self.yara_opsec_scan(file_path_on_disk)
@@ -613,12 +632,13 @@ class FileProcessor(TaskInterface):

                if alert_rules:
                    rule_matches_str = ", ".join(alert_rules)
-                    await self.alerter.file_data_alert(
-                        file_data=file_data,
-                        metadata=metadata,
-                        title="Yara rule match(es)",
-                        text=f"*Rules:* {rule_matches_str}",
-                    )
+                    if not file_previously_processed:
+                        await self.alerter.file_data_alert(
+                            file_data=file_data,
+                            metadata=metadata,
+                            title="Yara rule match(es)",
+                            text=f"*Rules:* {rule_matches_str}",
+                        )

        # if the file isn't a binary file (or is a Chromium history file)
        #   run NoseyParker on it for anything we can find
@@ -629,11 +649,12 @@ class FileProcessor(TaskInterface):

                if noseyparker_output and len(noseyparker_output.rule_matches) > 0:
                    file_data.noseyparker.CopyFrom(noseyparker_output)
-                    await self.alerter.file_data_alert(
-                        file_data=file_data,
-                        title="NoseyParker Results",
-                        metadata=metadata,
-                    )
+                    if not file_previously_processed:
+                        await self.alerter.file_data_alert(
+                            file_data=file_data,
+                            title="NoseyParker Results",
+                            metadata=metadata,
+                        )
            except Exception as e:
                await logger.aexception(e, message="Noseyparker scanning failed")
                enrichments_failure.append(constants.E_NOSEYPARKER_SCAN)
@@ -672,33 +693,34 @@ class FileProcessor(TaskInterface):
                        enrichments_success.append(constants.E_DOTNET_ANALYSIS)
                        file_data.analysis.CopyFrom(dotnet_results["analysis"])

-                        try:
-                            if file_data.analysis.dotnet_analysis.has_deserialization:
-                                await self.alerter.file_data_alert(
-                                    file_data=file_data,
-                                    metadata=metadata,
-                                    title="Potential Deserialization Found",
-                                )
-                        except:
-                            pass
-                        try:
-                            if file_data.analysis.dotnet_analysis.has_cmd_execution:
-                                await self.alerter.file_data_alert(
-                                    file_data=file_data,
-                                    metadata=metadata,
-                                    title="Potential Command Execution Found",
-                                )
-                        except:
-                            pass
-                        try:
-                            if file_data.analysis.dotnet_analysis.has_remoting:
-                                await self.alerter.file_data_alert(
-                                    file_data=file_data,
-                                    metadata=metadata,
-                                    title="Potential Remoting Found",
-                                )
-                        except:
-                            pass
+                        if not file_previously_processed:
+                            try:
+                                if file_data.analysis.dotnet_analysis.has_deserialization:
+                                    await self.alerter.file_data_alert(
+                                        file_data=file_data,
+                                        metadata=metadata,
+                                        title="Potential Deserialization Found",
+                                    )
+                            except:
+                                pass
+                            try:
+                                if file_data.analysis.dotnet_analysis.has_cmd_execution:
+                                    await self.alerter.file_data_alert(
+                                        file_data=file_data,
+                                        metadata=metadata,
+                                        title="Potential Command Execution Found",
+                                    )
+                            except:
+                                pass
+                            try:
+                                if file_data.analysis.dotnet_analysis.has_remoting:
+                                    await self.alerter.file_data_alert(
+                                        file_data=file_data,
+                                        metadata=metadata,
+                                        title="Potential Remoting Found",
+                                    )
+                            except:
+                                pass

                        if "object_id" in dotnet_results["decompilation"] and dotnet_results["decompilation"]["object_id"] is not None:
                            file_data.extracted_source = dotnet_results["decompilation"]["object_id"]
@@ -707,11 +729,12 @@ class FileProcessor(TaskInterface):
                                noseyparker_output = helpers.run_noseyparker_on_archive(temp_decomp_file.name)
                                if noseyparker_output:
                                    file_data.noseyparker.CopyFrom(noseyparker_output)
-                                    await self.alerter.file_data_alert(
-                                        file_data=file_data,
-                                        title="NoseyParker Results",
-                                        metadata=metadata,
-                                    )
+                                    if not file_previously_processed:
+                                        await self.alerter.file_data_alert(
+                                            file_data=file_data,
+                                            title="NoseyParker Results",
+                                            metadata=metadata,
+                                        )

                else:
                    enrichments_failure.append(constants.E_DOTNET_ANALYSIS)
@@ -404,14 +404,17 @@ class PostgresConnector(TaskInterface):

        for data in event.data:
            path = data.path
-            if not path.endswith("\\"):
-                path = f"{path}\\"

-            if re.match("^[a-zA-Z]{1}:\\.*", path):
+            if path.contains("\\") and not path.endswith("\\"):
+                path = f"{path}\\"
+            elif path.contains("/") and not path.endswith("/"):
+                path = f"{path}/"
+
+            if re.match(r"^([a-zA-Z]{1}:){0,1}[\\\/].*", path):
                # this is a file system path

                for item in data.items:
-                    if item.endswith("\\"):
+                    if item.endswith("\\") or item.endswith("/"):
                        object_type = "folder"
                    else:
                        object_type = "file"
@@ -456,7 +459,7 @@ class PostgresConnector(TaskInterface):
            else:
                extension = ""

-            if re.match(r"^[a-zA-Z]{1}:[\\/].*", path):
+            if re.match(r"^([a-zA-Z]{1}:){0,1}[\\\/].*", path):
                # this is a file system path

                f = FileInfo(
@@ -10,6 +10,8 @@ You could probably do 3 processors and 10 GB RAM, just might need to change how

 Additionally, only x64 architecture has been tested and is supported. ARM platforms (e.g., Mac devives with M* chips) are not currently supported but we intend to support these in the future.

+**Do not install the following requirements as root! Minikube is particular does not like to be run as root.**
+
 # Software Requirements
 **The following requirements need to be installed:**

@@ -24,6 +26,7 @@ Docker and docker-compose
 ```bash
 sudo apt-get update
 sudo apt-get install curl
+sudo mkdir /etc/apt/keyrings/ 2>/dev/null
 curl -fsSL https://get.docker.com -o get-docker.sh
 sudo sh get-docker.sh

@@ -26,9 +26,6 @@ kind: Ingress
 metadata:
  name: elastic-ingress
  annotations:
-    nginx.ingress.kubernetes.io/auth-type: basic
-    nginx.ingress.kubernetes.io/auth-secret: basic-auth
-    nginx.ingress.kubernetes.io/auth-realm: 'Authentication Required'
    ingress.kubernetes.io/ssl-redirect: "false"
    nginx.ingress.kubernetes.io/rewrite-target: /$2
    nginx.ingress.kubernetes.io/configuration-snippet: |
@@ -20,27 +20,18 @@ ch.setFormatter(formatter)
 logger.addHandler(ch)


-# Check dependencies
-exit_early = False
-try:
-    import boto3
-except:
-    logger.error("Please run `pip3 install boto3`")
-    exit_early = True
-try:
-    from vyper import v
-except:
-    logger.error("Please run `pip3 install vyper-config`")
-    exit_early = True
-
-try:
-    from passlib.hash import apr_md5_crypt
-except:
-    logger.error("Please run `pip3 install passlib`")
-    exit_early = True
-if exit_early:
+if os.geteuid() == 0:
+    logger.error("Please do not run this script as root")
    sys.exit(1)

+# Check dependencies
+try:
+    import boto3
+    from passlib.hash import apr_md5_crypt
+    from vyper import v
+except:
+    logger.error("Please run `pip3 install boto3 vyper-config passlib`")
+    sys.exit(1)

 version = "v0.1.0a"

@@ -233,22 +224,22 @@ def get_kubectl_value(key):

    elif key == "rabbitmq_admin_user":
        return run_cmd(
-            "kubectl get secret rabbitmq-creds -o=go-template='{{index .data \"rabbitmq_admin_user\"}}' | base64 -d"
+            "kubectl get secret rabbitmq-creds -o=go-template='{{index .data \"rabbitmq-admin-user\"}}' | base64 -d"
        )

    elif key == "rabbitmq_admin_password":
        return run_cmd(
-            "kubectl get secret rabbmitmq-creds -o=go-template='{{index .data \"rabbitmq_admin_password\"}}' | base64 -d"
+            "kubectl get secret rabbitmq-creds -o=go-template='{{index .data \"rabbitmq-admin-password\"}}' | base64 -d"
        )

    elif key == "rabbitmq_connectionuri":
        return run_cmd(
-            "kubectl get secret rabbmitmq-creds -o=go-template='{{index .data \"rabbitmq_connectionuri\"}}' | base64 -d"
+            "kubectl get secret rabbitmq-creds -o=go-template='{{index .data \"rabbitmq-connectionuri\"}}' | base64 -d"
        )

    elif key == "rabbitmq_erlang_cookie":
        return run_cmd(
-            "kubectl get secret rabbmitmq-creds -o=go-template='{{index .data \"rabbitmq_erlang_cookie\"}}' | base64 -d"
+            "kubectl get secret rabbitmq-creds -o=go-template='{{index .data \"rabbitmq-erlang-cookie\"}}' | base64 -d"
        )

    else: