From d3e603e4fed01634e3e183b1a77d5fd9303f71ff Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 14:04:35 -0700 Subject: [PATCH 01/11] Fix for saving rabbitmq creds to kubectl -Fix for saving rabbitmq creds to kubectl --- nemesis-cli.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/nemesis-cli.py b/nemesis-cli.py index bb01fe9..052d2d2 100755 --- a/nemesis-cli.py +++ b/nemesis-cli.py @@ -233,22 +233,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: From bcbd56dc336ad224948c93800898dc68361acee4 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 18:48:20 -0700 Subject: [PATCH 02/11] Removed double basic auth from Elasticsearch route -Removed ingress basic auth from Elasticsearch route as the route already used basic auth --- kubernetes/elastic/ingress.yaml | 3 --- 1 file changed, 3 deletions(-) diff --git a/kubernetes/elastic/ingress.yaml b/kubernetes/elastic/ingress.yaml index 5776d24..85bf316 100644 --- a/kubernetes/elastic/ingress.yaml +++ b/kubernetes/elastic/ingress.yaml @@ -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: | From 5dbfd797e503ea5d508b3339d56b691786d0fb0a Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 18:50:47 -0700 Subject: [PATCH 03/11] `mythic-sync` fixes -Fixed README.md and example settings.env file to contain all of the required settings -Bumped the mythic pypi package to the most recent version -More auth/connection error handling in the min sync code --- cmd/connectors/mythic-connector/LICENSE | 2 +- cmd/connectors/mythic-connector/README.md | 13 +- .../mythic-connector/requirements.txt | 3 +- cmd/connectors/mythic-connector/settings.env | 12 +- cmd/connectors/mythic-connector/sync.py | 165 ++++++++++++------ 5 files changed, 126 insertions(+), 69 deletions(-) diff --git a/cmd/connectors/mythic-connector/LICENSE b/cmd/connectors/mythic-connector/LICENSE index 5cccd45..b52ff06 100644 --- a/cmd/connectors/mythic-connector/LICENSE +++ b/cmd/connectors/mythic-connector/LICENSE @@ -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 diff --git a/cmd/connectors/mythic-connector/README.md b/cmd/connectors/mythic-connector/README.md index 629fc3b..b22ec56 100644 --- a/cmd/connectors/mythic-connector/README.md +++ b/cmd/connectors/mythic-connector/README.md @@ -15,30 +15,29 @@ 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 ``` +**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. - -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. +`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. ## References diff --git a/cmd/connectors/mythic-connector/requirements.txt b/cmd/connectors/mythic-connector/requirements.txt index d6416eb..cda82eb 100644 --- a/cmd/connectors/mythic-connector/requirements.txt +++ b/cmd/connectors/mythic-connector/requirements.txt @@ -1,6 +1,5 @@ aiohttp==3.8.1 redis==3.5.3 -#mythic==0.0.37 -mythic==0.1.0rc2 +mythic==0.1.4 requests==2.28.1 elasticsearch==8.4.1 \ No newline at end of file diff --git a/cmd/connectors/mythic-connector/settings.env b/cmd/connectors/mythic-connector/settings.env index 8bf2243..7af43f9 100644 --- a/cmd/connectors/mythic-connector/settings.env +++ b/cmd/connectors/mythic-connector/settings.env @@ -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 \ No newline at end of file +ELASTICSEARCH_USER=nemesis +ELASTICSEARCH_PASSWORD=password +MAX_FILE_SIZE=100000000 +EXPIRATION_DAYS=100 \ No newline at end of file diff --git a/cmd/connectors/mythic-connector/sync.py b/cmd/connectors/mythic-connector/sync.py index 7827f0d..defaf51 100644 --- a/cmd/connectors/mythic-connector/sync.py +++ b/cmd/connectors/mythic-connector/sync.py @@ -12,7 +12,7 @@ from datetime import datetime, timedelta import aiohttp 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 @@ -272,8 +272,6 @@ async def handle_file(mythic_instance: mythic_classes.Mythic) -> None: 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}}) { @@ -752,36 +750,50 @@ 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 + 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 +801,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,23 +822,17 @@ async def wait_for_authentication() -> mythic_classes.Mythic: server_ip=MYTHIC_IP, server_port=MYTHIC_PORT, ssl=True, - timeout=-1, - ) - except Exception: - mythic_sync_log.exception( - "Encountered an exception while trying to authenticate to Mythic, trying again in %s seconds...", - WAIT_TIMEOUT, - ) - 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: + if hasattr(e, 'errors') and len(e.errors) > 0 and "message" in e.errors[0]: + message = e.errors[0]["message"] + mythic_sync_log.error( + f"Encountered an exception while trying to authenticate to Mythic, trying again in {WAIT_TIMEOUT} seconds: '{message}'" + ) + else: + 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 elif MYTHIC_USERNAME == "" and MYTHIC_PASSWORD == "": @@ -861,32 +873,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) From df97ff0901c4b53a73db55ef5126815c979ee11a Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 19:00:50 -0700 Subject: [PATCH 04/11] Added check to ensure `./nemesis-cli.py` is not run as root -Added check to ensure `./nemesis-cli.py` is not run as root --- docs/setup.md | 2 ++ nemesis-cli.py | 29 ++++++++++------------------- 2 files changed, 12 insertions(+), 19 deletions(-) diff --git a/docs/setup.md b/docs/setup.md index f065cbf..ed100af 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -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:** diff --git a/nemesis-cli.py b/nemesis-cli.py index 052d2d2..878113e 100755 --- a/nemesis-cli.py +++ b/nemesis-cli.py @@ -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" From 714922f8a2cddde320188b19dcc412a8e049aa00 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 19:03:09 -0700 Subject: [PATCH 05/11] Update setup.md --- docs/setup.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/setup.md b/docs/setup.md index ed100af..0e80f34 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -26,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 From 35e6407280d6deb3ba96c89d3028ea05f37a85d1 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 19:05:35 -0700 Subject: [PATCH 06/11] Update README.md --- cmd/connectors/mythic-connector/README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cmd/connectors/mythic-connector/README.md b/cmd/connectors/mythic-connector/README.md index b22ec56..0a58f33 100644 --- a/cmd/connectors/mythic-connector/README.md +++ b/cmd/connectors/mythic-connector/README.md @@ -23,6 +23,8 @@ 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`: From 7a20dbd51e3bd378c7f6e25ff1ddf1f13ce2c400 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 19:49:55 -0700 Subject: [PATCH 07/11] Update README.md's main skaffold launch command -Update README.md's main skaffold launch command --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 27d2b5a..dd8ace2 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,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. From dd831f2c8cef0943919a3bb6e24f7aa35c45ad45 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Tue, 22 Aug 2023 22:04:48 -0700 Subject: [PATCH 08/11] Updated README.md with relevant blog posts -Updated README.md with relevant blog posts --- README.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/README.md b/README.md index dd8ace2..d90c2da 100644 --- a/README.md +++ b/README.md @@ -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) From e114b8fa429db0281533a785a71107ec246f0fa7 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Thu, 24 Aug 2023 18:26:02 -0700 Subject: [PATCH 09/11] Various fixes for `mythic-connector` -Additional error handling -Fixes for file browser handling -Re-enabled file process/process handling -Owner pulled from Linux file permissions -Lots of misc fixes -Implemented CLEAR_REDIS in an env variable to reprocess data --- cmd/connectors/mythic-connector/README.md | 4 + .../mythic-connector/requirements.txt | 5 +- cmd/connectors/mythic-connector/sync.py | 315 +++++++++++------- 3 files changed, 194 insertions(+), 130 deletions(-) diff --git a/cmd/connectors/mythic-connector/README.md b/cmd/connectors/mythic-connector/README.md index 0a58f33..9432026 100644 --- a/cmd/connectors/mythic-connector/README.md +++ b/cmd/connectors/mythic-connector/README.md @@ -41,6 +41,10 @@ Ensure the host where `mythic_nemesis_sync` is running has network access to the `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. +## 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 - [Mythic](https://github.com/its-a-feature/Mythic) - Multi-platform C2 Framework diff --git a/cmd/connectors/mythic-connector/requirements.txt b/cmd/connectors/mythic-connector/requirements.txt index cda82eb..decd119 100644 --- a/cmd/connectors/mythic-connector/requirements.txt +++ b/cmd/connectors/mythic-connector/requirements.txt @@ -1,5 +1,6 @@ aiohttp==3.8.1 redis==3.5.3 -mythic==0.1.4 +mythic==0.1.5 requests==2.28.1 -elasticsearch==8.4.1 \ No newline at end of file +elasticsearch==8.4.1 +gql==3.4.1 \ No newline at end of file diff --git a/cmd/connectors/mythic-connector/sync.py b/cmd/connectors/mythic-connector/sync.py index defaf51..fcccbb4 100644 --- a/cmd/connectors/mythic-connector/sync.py +++ b/cmd/connectors/mythic-connector/sync.py @@ -10,6 +10,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta import aiohttp +import gql import redis import requests from elasticsearch import AuthenticationException, Elasticsearch @@ -18,7 +19,10 @@ from elasticsearch import AuthenticationException, Elasticsearch 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,10 +278,6 @@ 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: @@ -298,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] @@ -358,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( @@ -472,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 } } @@ -479,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 @@ -490,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"] @@ -505,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"] @@ -517,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 @@ -545,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: @@ -559,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: @@ -588,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 @@ -664,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 @@ -675,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] @@ -786,6 +832,11 @@ async def wait_for_redis() -> None: 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) + 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( @@ -822,17 +873,19 @@ async def wait_for_authentication() -> mythic_classes.Mythic: server_ip=MYTHIC_IP, server_port=MYTHIC_PORT, ssl=True, + logging_level=logging.ERROR, + timeout=-1 ) + 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 except Exception as e: - if hasattr(e, 'errors') and len(e.errors) > 0 and "message" in e.errors[0]: - message = e.errors[0]["message"] - mythic_sync_log.error( - f"Encountered an exception while trying to authenticate to Mythic, trying again in {WAIT_TIMEOUT} seconds: '{message}'" - ) - else: - mythic_sync_log.error( - f"Encountered an exception while trying to authenticate to Mythic, trying again in {WAIT_TIMEOUT} seconds: {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 elif MYTHIC_USERNAME == "" and MYTHIC_PASSWORD == "": @@ -856,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 @@ -952,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()) From b3487912e93f02d50ac627f63bde2dac3a4f2a0c Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Thu, 24 Aug 2023 18:28:15 -0700 Subject: [PATCH 10/11] Fixed path processing for file listings in the postgres connector -Fixed path processing for file listings in the postgres connector --- .../tasks/postgres_connector/postgres_connector.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/cmd/enrichment/enrichment/tasks/postgres_connector/postgres_connector.py b/cmd/enrichment/enrichment/tasks/postgres_connector/postgres_connector.py index eb252c8..503be95 100644 --- a/cmd/enrichment/enrichment/tasks/postgres_connector/postgres_connector.py +++ b/cmd/enrichment/enrichment/tasks/postgres_connector/postgres_connector.py @@ -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( From 0a08cd9497bd6a35ef51b9a090f4130ec5f79f91 Mon Sep 17 00:00:00 2001 From: HarmJ0y Date: Fri, 25 Aug 2023 13:41:05 -0700 Subject: [PATCH 11/11] Don't alert on files that have already been processed -Don't alert on files that have already been processed --- cmd/enrichment/enrichment/containers.py | 3 +- cmd/enrichment/enrichment/lib/nemesis_db.py | 13 ++ .../enrichment/tasks/file_processor.py | 155 ++++++++++-------- 3 files changed, 104 insertions(+), 67 deletions(-) diff --git a/cmd/enrichment/enrichment/containers.py b/cmd/enrichment/enrichment/containers.py index 55c7471..aed032f 100644 --- a/cmd/enrichment/enrichment/containers.py +++ b/cmd/enrichment/enrichment/containers.py @@ -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, diff --git a/cmd/enrichment/enrichment/lib/nemesis_db.py b/cmd/enrichment/enrichment/lib/nemesis_db.py index 849eb74..08fa252 100644 --- a/cmd/enrichment/enrichment/lib/nemesis_db.py +++ b/cmd/enrichment/enrichment/lib/nemesis_db.py @@ -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 diff --git a/cmd/enrichment/enrichment/tasks/file_processor.py b/cmd/enrichment/enrichment/tasks/file_processor.py index e4dbae6..c58deaa 100644 --- a/cmd/enrichment/enrichment/tasks/file_processor.py +++ b/cmd/enrichment/enrichment/tasks/file_processor.py @@ -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)