diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..8ae9528 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +.git +.github +.DS_Store +__pycache__ +*.pyc +test_sync.py diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..8ba1e84 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,14 @@ +version: 2 +updates: + - package-ecosystem: pip + directory: / + schedule: + interval: monthly + - package-ecosystem: docker + directory: / + schedule: + interval: monthly + - package-ecosystem: github-actions + directory: / + schedule: + interval: monthly diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..4d1a35f --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,61 @@ +name: Test + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +jobs: + unit-tests: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: python -m pip install --upgrade pip + - run: python -m pip install -r requirements.txt + - run: python -m unittest -v + - run: python -m py_compile sync.py test_sync.py + + container-build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: docker/setup-buildx-action@v3 + - uses: docker/build-push-action@v6 + with: + context: . + load: true + push: false + tags: mythic-sync:test + - name: Verify Redis AOF persistence across container recreation + run: | + docker volume create mythic_sync_test_redis + docker run --detach --name mythic-sync-redis-before \ + --volume mythic_sync_test_redis:/data \ + --entrypoint redis-server mythic-sync:test \ + --appendonly yes --appendfsync always --dir /data + until docker exec mythic-sync-redis-before redis-cli ping; do sleep 1; done + docker exec mythic-sync-redis-before redis-cli set mythic_sync:persistence_test queued + docker stop mythic-sync-redis-before + docker rm mythic-sync-redis-before + docker run --detach --name mythic-sync-redis-after \ + --volume mythic_sync_test_redis:/data \ + --entrypoint redis-server mythic-sync:test \ + --appendonly yes --appendfsync always --dir /data + until docker exec mythic-sync-redis-after redis-cli ping; do sleep 1; done + test "$(docker exec mythic-sync-redis-after redis-cli get mythic_sync:persistence_test)" = queued + - name: Clean up Redis persistence test + if: always() + run: | + docker rm --force mythic-sync-redis-before mythic-sync-redis-after 2>/dev/null || true + docker volume rm mythic_sync_test_redis 2>/dev/null || true diff --git a/.gitignore b/.gitignore index bc8a670..27216c0 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,4 @@ -.idea/* \ No newline at end of file +.idea/* +.DS_Store +__pycache__/ +*.py[cod] diff --git a/CHANGELOG.md b/CHANGELOG.md index ace7c78..64ad60e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,51 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +## [3.1.0] - 18 July 2026 + +### Added + +* Added an AOF-backed Redis retry queue so Ghostwriter tag failures do not block log entry ingestion and pending jobs survive `mythic_sync` restarts and service recreation, provided the Redis data volume is preserved. +* Added Redis health checks, pending-job reporting, and automatic migration of legacy mappings and queued tag jobs. +* Added tests for query retries, stale entry reconciliation, deleted entry recreation, tag retries, and source IP formatting. +* Added a container-level CI check that verifies queued Redis data survives a Redis container restart. +* Added startup warnings for embedded Redis volume requirements and actionable dead-letter inspection details. +* Added optional `REDIS_URL` support for authenticated and TLS-protected external Redis services without logging URL credentials. +* Added GitHub Actions coverage for Python 3.10, Python 3.12, and production container builds, plus Dependabot configuration. + +### Changed + +* Changed Ghostwriter GraphQL retries to use exponential backoff with jitter, capped at five minutes. +* Improved GraphQL error messages with the operation name and variables, including actionable context for ambiguous `ModelDoesNotExist` responses. +* Changed source IP formatting from a JSON array to a sorted, comma-separated string. +* Made Redis hostname, port, and database configurable and scoped Redis mappings by Mythic host, Mythic port, and Ghostwriter oplog. +* Made `MYTHIC_PORT` optional and defaulted it to Mythic's standard HTTPS port, `7443`, preserving existing install behavior. +* Enabled Redis append-only persistence for embedded Mythic deployments and for the standalone Compose Redis service with a named volume. Raw Mythic subscription events remain live-streamed and are not persisted locally. +* Changed Redis AOF flushing to `appendfsync always` and added graceful embedded Redis shutdown handling. +* Updated supported dependency pins and moved the production container to Python 3.11 on Debian Bookworm. +* Made Mythic timestamp and IP parsing tolerant of timezone variants, IPv6, plain strings, CIDR notation, and malformed individual addresses. +* Filtered loopback, unspecified, multicast, and IPv6 link-local source addresses while preserving potentially useful private, CGNAT, ULA, and global addresses. +* Limited GraphQL error context to identifiers while redacting commands, comments, and other potentially sensitive values. +* Made the Ghostwriter initialization entry idempotent and corrected Mythic API-key authentication so user credentials are not also required. +* Made subscription and tag worker shutdown explicit so worker failures cancel and await the remaining tasks before the Ghostwriter client closes. +* Documented dead-letter inspection, the single-replica constraint, and a manual Mythic/Ghostwriter acceptance checklist. + +### Fixed + +* Fixed stale Redis entry mappings by looking up the Ghostwriter entry by `entryIdentifier` and repairing the mapping or recreating a deleted entry. +* Fixed `ModelDoesNotExist` update failures retrying the same stale ID forever instead of reconciling by `entryIdentifier`. +* Fixed pending tag jobs targeting deleted entries retrying forever; jobs now move to a replacement entry or enter a durable Redis dead-letter hash with an operator notification. +* Prevented repeated Mythic error notifications and notification delivery failures from interfering with GraphQL retries. +* Fixed Redis startup checks reporting success without issuing a command. +* Prevented Redis startup failures from attempting Mythic notifications before Mythic authentication is established. +* Ensured task cancellation exits Redis, service, authentication, GraphQL, and tag retry loops instead of being handled as a retryable failure. +* Prevented Mythic instances sharing an IP but using different ports from colliding in Redis state or initialization entry identifiers; existing IP-only Redis state migrates on startup. +* Stopped the standalone Compose file from overriding Redis host, port, and database values supplied through `settings.env`. +* Fixed conversion, creation, update, and Redis failures being swallowed after logging, which could allow processing to continue after an entry failed. +* Fixed timezone-aware token expiration parsing for timestamps ending in `Z`. + ## [3.0.8] - 25 July 2025 ### Changed diff --git a/Dockerfile b/Dockerfile index a87d7ff..50edd70 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,11 +1,22 @@ -FROM redis:7-bullseye +FROM redis:7-bookworm -RUN apt update && apt install python3 python3-pip -y \ - --no-install-recommends +RUN apt-get update \ + && apt-get install -y --no-install-recommends python3 python3-venv \ + && rm -rf /var/lib/apt/lists/* -COPY requirements.txt . -RUN python3 -m pip install -r requirements.txt +WORKDIR /app -COPY sync.py . +COPY requirements.txt ./ +RUN python3 -m venv /opt/venv \ + && /opt/venv/bin/pip install --no-cache-dir -r requirements.txt -CMD ["bash", "-c", "redis-server & python3 -u sync.py"] +COPY sync.py docker-entrypoint.sh ./ +RUN chmod +x /app/docker-entrypoint.sh + +VOLUME ["/data"] + +STOPSIGNAL SIGTERM + +USER redis + +ENTRYPOINT ["/app/docker-entrypoint.sh"] diff --git a/README.md b/README.md index bc0935b..abc996c 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ For the easiest experience with `mythic_sync`, install it via the `mythic-cli` t On your Mythic server, run: `sudo ./mythic-cli mythic_sync install github https://github.com/GhostManager/mythic_sync` Follow the prompts to configure `mythic_sync` with your Mythic and Ghostwriter server configuration. +If `MYTHIC_PORT` is not provided, `mythic_sync` uses Mythic's default HTTPS port, `7443`. You can get your Ghostwriter Oplog ID by visiting your log in your web browser and looking at the top of the page or the URL. A URL with `/oplog/12/entries` means your Oplog ID is `12`. @@ -61,11 +62,39 @@ After cloning repository, open the `settings.env` file and fill in the variables ```text MYTHIC_IP=10.10.1.100 +MYTHIC_PORT=7443 MYTHIC_USERNAME=mythic_admin MYTHIC_PASSWORD=SuperSecretPassword -GHOSTWRITER_API_KEY=eyJ0eXAiO... +GHOSTWRITER_API_KEY=gwst_... GHOSTWRITER_URL=https://ghostwriter.mydomain.com GHOSTWRITER_OPLOG_ID=12 +REDIS_HOSTNAME=redis +REDIS_PORT=6379 +REDIS_DB=1 +``` + +The standalone Compose deployment reads these Redis settings directly from `settings.env`; edit +that file to select a different Redis host, port, or database without modifying +`docker-compose.yml`. + +Set `MYTHIC_API_KEY` to authenticate with a Mythic API key instead of `MYTHIC_USERNAME` and +`MYTHIC_PASSWORD`. `MYTHIC_PORT` is optional and defaults to `7443`. + +The standalone Compose deployment stores entry mappings and pending tag updates in the named +`mythic_sync_redis` volume. Redis append-only persistence with `appendfsync always` is enabled so recreating the +`mythic_sync` application container does not discard pending work. Mythic installations default +to an embedded append-only Redis instance for compatibility. To preserve that data across service +recreation, configure Mythic's generated Compose service to mount a stable named volume at `/data`; +the Dockerfile's volume declaration alone does not guarantee that a replacement container will +reattach the same storage. Alternatively, `REDIS_HOSTNAME`, `REDIS_PORT`, and `REDIS_DB` can select +an external persistent Redis instance. + +For an authenticated or TLS-protected external Redis service, set `REDIS_URL` instead of the three +individual Redis settings. The URL takes precedence and may include the database number, username, +and password. Its credentials are never written to the service log: + +```text +REDIS_URL=rediss://sync-user:password@redis.example.com:6380/1 ``` Once the environment variables are set up, you can launch the service by using `docker-compose`: @@ -78,7 +107,7 @@ docker-compose up Open your Ghostwriter log and look for an initial entry. You should see something like the following: - > Initial entry from mythic_sync at: . If you're seeing this then oplog syncing is working for this C2 server! + > Initial entry from mythic_sync at: :. If you're seeing this then oplog syncing is working for this C2 server! If so, you're all set! Otherwise, check the logs from the docker container for error messages. Fetch the logs with: @@ -88,9 +117,85 @@ If so, you're all set! Otherwise, check the logs from the docker container for e Ensure the host where `mythic_sync` is running has network access to the Ghostwriter and Mythic servers. -`mythic_sync` uses an internal Redis database to sync what events have already been sent to Ghostwriter, avoiding duplicates. +`mythic_sync` uses Redis to track events already sent to Ghostwriter and to retain pending tag +updates. Redis mappings are scoped by Mythic host, Mythic port, and Ghostwriter oplog. Legacy +mappings and pending tag jobs are migrated automatically when first accessed after an upgrade. +Migration from the earlier IP-only namespace assumes the old state belongs to the first upgraded +instance; state that was already mixed by multiple Mythic ports cannot be separated automatically. + +Source IP normalization removes duplicate, loopback, unspecified, multicast, and IPv6 link-local +addresses to keep multi-adapter hosts readable. Private IPv4, CGNAT, IPv6 ULA, and globally routable +addresses are preserved because interface metadata is not available to identify container bridges +reliably. +If a queued tag job's entry no longer resolves by `entryIdentifier`, the worker removes it from +active retries but preserves its payload in the deployment's namespaced Redis dead-letter hash and +notifies Mythic. A later task update can create a fresh tag job for the current entry. + +The dead-letter hash is named: + +```text +mythic_sync::::pending_tag_updates:dead_letter +``` + +The startup warning prints the exact key. For the standalone Compose deployment, inspect it with: + +```bash +docker compose exec redis redis-cli -n 1 HGETALL \ + "mythic_sync::::pending_tag_updates:dead_letter" +``` + +After investigating an individual job, remove only that field with `HDEL `. +There is intentionally no automatic dead-letter replay in this release; updating the corresponding +Mythic task creates a fresh tag job if Ghostwriter can resolve or recreate its oplog entry. + +The embedded Redis process flushes and shuts down when the application container receives +`SIGTERM`. Both embedded and standalone configurations use synchronous AOF writes. Do not remove +or renew the Redis data volume during upgrades if queued work must survive. + +At startup, the service logs the selected Redis endpoint and number of pending tag updates. A +successful Redis client construction is not sufficient: the service waits for `PING` to succeed +before subscribing to Mythic events. + +Run only one `mythic_sync` instance for a given Mythic host, port, and Ghostwriter oplog namespace. +This release does not coordinate tag workers across multiple active replicas. + +### Manual Acceptance Checklist + +Perform destructive checks only in a disposable test oplog. + +1. Start `mythic_sync` and confirm the initialization entry appears once in Ghostwriter. Restart + the service and confirm a duplicate initialization entry is not created. +2. Run a Mythic task and confirm Ghostwriter receives an oplog entry whose `entryIdentifier` + matches the Mythic `agent_task_id`. Update the task and confirm the existing entry changes rather + than creating a duplicate. +3. Delete that Ghostwriter entry, then update the Mythic task. Confirm the logs identify the stale + Ghostwriter ID and `entryIdentifier`, and that the entry is found again or recreated. +4. Add, change, and remove tags on the Mythic task. Confirm Ghostwriter's corresponding `mythic:` + tags converge to the current Mythic tags while any Ghostwriter tags without the `mythic:` prefix + remain unchanged. Stale-target and dead-letter behavior is covered by the automated test suite. +5. With a stable Redis volume attached, note the startup pending/dead-letter counts, restart or + recreate only the `mythic_sync` container, and confirm the counts and Redis data remain. Do not + remove or renew the volume during this check. +6. Temporarily use a service token without the required test-oplog permission and confirm the error + identifies the GraphQL operation, oplog or entry identifier, and likely permission problem without + logging command or comment contents. + +### Testing + +Run the unit suite in the production dependency image: + +```bash +docker build -t mythic-sync-test . +docker run --rm --entrypoint /opt/venv/bin/python \ + -e PYTHONDONTWRITEBYTECODE=1 \ + -v "$PWD:/workspace" -w /workspace \ + mythic-sync-test -m unittest -v +``` -If the `mythic_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. +The suite covers retries, `ModelDoesNotExist` reconciliation, stale and deleted entries, Redis +migrations, tag queue processing, +authentication selection, timestamp and IP normalization, initialization idempotency, and +diagnostic redaction. ## References diff --git a/docker-compose.yml b/docker-compose.yml index 0830873..0529f1e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,10 +1,24 @@ -version: "3" services: mythic_sync: build: . depends_on: - - redis + redis: + condition: service_healthy env_file: - settings.env + restart: unless-stopped redis: - image: redis:5-alpine + image: redis:7-alpine + command: ["redis-server", "--appendonly", "yes", "--appendfsync", "always"] + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 10 + restart: unless-stopped + stop_grace_period: 30s + volumes: + - mythic_sync_redis:/data + +volumes: + mythic_sync_redis: diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh new file mode 100644 index 0000000..109a1be --- /dev/null +++ b/docker-entrypoint.sh @@ -0,0 +1,28 @@ +#!/bin/sh +set -eu + +redis_hostname="${REDIS_HOSTNAME:-127.0.0.1}" +redis_url="${REDIS_URL:-}" + +if [ -z "$redis_url" ] && { [ "$redis_hostname" = "127.0.0.1" ] || [ "$redis_hostname" = "localhost" ]; }; then + redis-server --appendonly yes --appendfsync always --dir /data --daemonize yes + + /opt/venv/bin/python -u /app/sync.py & + app_pid=$! + + shutdown() { + trap - TERM INT + kill -TERM "$app_pid" 2>/dev/null || true + redis-cli -h 127.0.0.1 shutdown save 2>/dev/null || true + } + + trap shutdown TERM INT + set +e + wait "$app_pid" + app_status=$? + set -e + redis-cli -h 127.0.0.1 shutdown save 2>/dev/null || true + exit "$app_status" +fi + +exec /opt/venv/bin/python -u /app/sync.py diff --git a/requirements.txt b/requirements.txt index 9a7220e..ea56533 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -aiohttp==3.8.4 -redis==4.5.4 -mythic==0.2.6 -gql==3.4.0 +aiohttp==3.14.1 +redis==8.0.1 +mythic==0.2.10 +gql==3.5.3 diff --git a/settings.env b/settings.env index 0d1d81a..b4fa047 100644 --- a/settings.env +++ b/settings.env @@ -7,3 +7,5 @@ GHOSTWRITER_URL=https://ghostwriter.mydomain.com GHOSTWRITER_OPLOG_ID=123 REDIS_HOSTNAME=redis REDIS_PORT=6379 +REDIS_DB=1 +# REDIS_URL=rediss://username:password@redis.example.com:6380/1 diff --git a/sync.py b/sync.py index 6864e48..4574866 100644 --- a/sync.py +++ b/sync.py @@ -1,10 +1,13 @@ # Standard Libraries import asyncio +import builtins import ipaddress import json import logging import os +import random import sys +import time from asyncio.exceptions import TimeoutError from datetime import datetime, timedelta, timezone @@ -20,7 +23,7 @@ # Mythic Sync Libraries from mythic import mythic, mythic_classes -VERSION = "3.0.8" +VERSION = "3.1.0" # Logging configuration # Level applies to all loggers, including ``gql`` Transport and Client loggers @@ -49,6 +52,21 @@ class MythicSync: # How long to wait for a service to start before retrying an HTTP request wait_timeout = 5 + max_retry_timeout = 300 + + # Polling interval for independently retried tag updates + tag_retry_poll_interval = 5 + legacy_tag_retry_data_key = "mythic_sync:pending_tag_updates:data" + legacy_tag_retry_schedule_key = "mythic_sync:pending_tag_updates:schedule" + + # Variables which identify a failed query without exposing commands or other sensitive content + diagnostic_variable_names = { + "entry_identifier", + "id", + "oplog", + "oplog_entry_id", + "oplogId", + } # Query for the whoami expiration checks whoami_query = gql( @@ -75,12 +93,16 @@ class MythicSync: # Query for the first log sent after initialization initial_query = gql( """ - mutation InitializeMythicSync ($oplogId: bigint!, $description: String!, $server: String!, $extraFields: jsonb!) { + mutation InitializeMythicSync ( + $oplogId: bigint!, $description: String!, $server: String!, $entry_identifier: String!, + $extraFields: jsonb! + ) { insert_oplogEntry(objects: { oplog: $oplogId, description: $description, sourceIp: $server, tool: "Mythic", + entryIdentifier: $entry_identifier, extraFields: $extraFields }) { returning { id } @@ -183,18 +205,15 @@ class MythicSync: mythic_sync_log.error("MYTHIC_IP must be supplied!") sys.exit(1) - MYTHIC_PORT = os.environ.get("MYTHIC_PORT") - if MYTHIC_PORT is None: - mythic_sync_log.error("MYTHIC_PORT must be supplied!") - sys.exit(1) - else: - MYTHIC_PORT = int(MYTHIC_PORT) + MYTHIC_PORT = int(os.environ.get("MYTHIC_PORT") or "7443") MYTHIC_URL = f"https://{MYTHIC_IP}:{MYTHIC_PORT}" # Redis server - REDIS_HOSTNAME = "127.0.0.1" - REDIS_PORT = 6379 + REDIS_URL = os.environ.get("REDIS_URL") or "" + REDIS_HOSTNAME = os.environ.get("REDIS_HOSTNAME", "127.0.0.1") + REDIS_PORT = int(os.environ.get("REDIS_PORT", "6379")) + REDIS_DB = int(os.environ.get("REDIS_DB", "1")) # Ghostwriter server authentication GHOSTWRITER_API_KEY = os.environ.get("GHOSTWRITER_API_KEY") @@ -220,14 +239,27 @@ class MythicSync: "Authorization": f"Bearer {GHOSTWRITER_API_KEY}", "Content-Type": "application/json" } - last_error_timestamp = datetime.utcnow() - timedelta(hours=1) last_error_delta = timedelta(minutes=30) session = None client = None transport = AIOHTTPTransport(url=GRAPHQL_URL, timeout=10, headers=headers) def __init__(self): - pass + self._notification_timestamps = {} + self.legacy_ip_redis_namespace = f"mythic_sync:{self.GHOSTWRITER_OPLOG_ID}:{self.MYTHIC_IP}" + self.redis_namespace = f"{self.legacy_ip_redis_namespace}:{self.MYTHIC_PORT}" + self.tag_retry_data_key = f"{self.redis_namespace}:pending_tag_updates:data" + self.tag_retry_schedule_key = f"{self.redis_namespace}:pending_tag_updates:schedule" + self.tag_retry_dead_letter_key = f"{self.redis_namespace}:pending_tag_updates:dead_letter" + self.legacy_ip_tag_retry_data_key = ( + f"{self.legacy_ip_redis_namespace}:pending_tag_updates:data" + ) + self.legacy_ip_tag_retry_schedule_key = ( + f"{self.legacy_ip_redis_namespace}:pending_tag_updates:schedule" + ) + self.legacy_ip_tag_retry_dead_letter_key = ( + f"{self.legacy_ip_redis_namespace}:pending_tag_updates:dead_letter" + ) async def initialize(self) -> None: """ @@ -249,21 +281,205 @@ async def initialize(self) -> None: await self._check_token() await self._create_initial_entry() - async def _get_sorted_ips(self, ip: str) -> str: - source_ips = json.loads(ip) - # account for CIDR notation (ex: 192.168.0.123/24) in IPs list to make sure we only get the actual IP - source_ips = [x.split("/")[0] for x in source_ips if x != ""] - source_ipv4 = [] - for i in range(len(source_ips)): - new_address = ipaddress.ip_address(source_ips[i]) - if isinstance(new_address, ipaddress.IPv4Address): - source_ipv4.append(new_address) - source_ipv4 = [str(x) for x in sorted(source_ipv4)] - source_ips = source_ipv4 - source_ip = json.dumps(source_ips) - return source_ip - - async def _execute_query(self, query: DocumentNode, variable_values: dict = None) -> dict: + async def _get_sorted_ips(self, ip: object) -> str: + """Normalize a Mythic IP value into a sorted IPv4/IPv6 string.""" + if not ip: + return "" + + source_ips = ip + if isinstance(ip, str): + try: + source_ips = json.loads(ip) + except json.JSONDecodeError: + source_ips = [ip] + if isinstance(source_ips, str): + source_ips = [source_ips] + if not isinstance(source_ips, (list, tuple, set)): + raise ValueError(f"Unsupported Mythic IP value: {source_ips!r}") + + addresses = set() + omitted_addresses = 0 + for source_ip in source_ips: + source_ip = str(source_ip).strip() + if not source_ip: + continue + try: + address = ipaddress.ip_address(source_ip.split("/", 1)[0]) + if ( + address.is_loopback + or address.is_unspecified + or address.is_multicast + or (address.version == 6 and address.is_link_local) + ): + omitted_addresses += 1 + continue + addresses.add(address) + except ValueError: + mythic_sync_log.warning("Ignoring invalid IP address reported by Mythic: %r", source_ip) + + if omitted_addresses: + mythic_sync_log.debug( + "Omitted %s loopback, unspecified, multicast, or IPv6 link-local address(es) reported by Mythic", + omitted_addresses, + ) + + return ", ".join( + str(address) for address in sorted(addresses, key=lambda address: (address.version, int(address))) + ) + + @staticmethod + def _parse_mythic_timestamp(value: str) -> str: + """Normalize Mythic timestamps while retaining any timezone information.""" + if not value: + return "" + normalized = value[:-1] + "+00:00" if value.endswith("Z") else value + return datetime.fromisoformat(normalized).isoformat() + + def _redis_entry_key(self, entry_identifier: str) -> str: + """Return a Redis mapping key scoped to this Mythic server and Ghostwriter oplog.""" + return f"{self.redis_namespace}:entry:{entry_identifier}" + + def _legacy_ip_redis_entry_key(self, entry_identifier: str) -> str: + """Return the pre-port mapping key used by earlier 3.1.0 builds.""" + return f"{self.legacy_ip_redis_namespace}:entry:{entry_identifier}" + + def _get_cached_entry_id(self, entry_identifier: str): + """Read a scoped entry mapping, migrating the legacy raw key when encountered.""" + scoped_key = self._redis_entry_key(entry_identifier) + entry_id = self.rconn.get(scoped_key) + if entry_id is not None: + return entry_id + + for legacy_key in ( + self._legacy_ip_redis_entry_key(entry_identifier), + entry_identifier, + ): + legacy_entry_id = self.rconn.get(legacy_key) + if legacy_entry_id is not None: + pipeline = self.rconn.pipeline(transaction=True) + pipeline.set(scoped_key, legacy_entry_id) + pipeline.delete(legacy_key) + pipeline.execute() + mythic_sync_log.info("Migrated legacy Redis mapping for %s", entry_identifier) + return legacy_entry_id + return None + + def _set_cached_entry_id(self, entry_identifier: str, ghostwriter_entry_id: str) -> None: + self.rconn.set(self._redis_entry_key(entry_identifier), ghostwriter_entry_id) + + def _delete_cached_entry_id(self, entry_identifier: str) -> None: + pipeline = self.rconn.pipeline(transaction=True) + pipeline.delete(self._redis_entry_key(entry_identifier)) + pipeline.delete(self._legacy_ip_redis_entry_key(entry_identifier)) + pipeline.delete(entry_identifier) + pipeline.execute() + + def _migrate_tag_queue(self, data_key: str, schedule_key: str) -> int: + """Move one legacy pending-tag queue into the current port-scoped queue.""" + legacy_payloads = self.rconn.hgetall(data_key) + legacy_schedule = dict( + self.rconn.zrange(schedule_key, 0, -1, withscores=True) + ) + pipeline = self.rconn.pipeline(transaction=True) + migrated = 0 + for entry_id, payload in legacy_payloads.items(): + if self.rconn.hget(self.tag_retry_data_key, entry_id) is None: + pipeline.hset(self.tag_retry_data_key, entry_id, payload) + pipeline.zadd( + self.tag_retry_schedule_key, + {entry_id: legacy_schedule.get(entry_id, time.time())}, + ) + migrated += 1 + pipeline.delete(data_key) + pipeline.delete(schedule_key) + pipeline.execute() + return migrated + + def _migrate_dead_letters(self, dead_letter_key: str) -> int: + """Move one legacy dead-letter hash into the current port-scoped hash.""" + legacy_payloads = self.rconn.hgetall(dead_letter_key) + pipeline = self.rconn.pipeline(transaction=True) + migrated = 0 + for entry_id, payload in legacy_payloads.items(): + if self.rconn.hget(self.tag_retry_dead_letter_key, entry_id) is None: + pipeline.hset(self.tag_retry_dead_letter_key, entry_id, payload) + migrated += 1 + pipeline.delete(dead_letter_key) + pipeline.execute() + return migrated + + def _migrate_legacy_tag_queue(self) -> int: + """Move pre-namespace and pre-port tag state into this deployment's scoped keys.""" + migrated = self._migrate_tag_queue( + self.legacy_tag_retry_data_key, + self.legacy_tag_retry_schedule_key, + ) + + ip_scoped_count = self.rconn.hlen(self.legacy_ip_tag_retry_data_key) + ip_scoped_dead_letters = self.rconn.hlen(self.legacy_ip_tag_retry_dead_letter_key) + if ip_scoped_count or ip_scoped_dead_letters: + mythic_sync_log.warning( + "Migrating %s IP-only Redis tag record(s) to Mythic port %s; pre-port state cannot be " + "separated if multiple Mythic instances previously shared this IP and oplog", + ip_scoped_count + ip_scoped_dead_letters, + self.MYTHIC_PORT, + ) + migrated += self._migrate_tag_queue( + self.legacy_ip_tag_retry_data_key, + self.legacy_ip_tag_retry_schedule_key, + ) + migrated += self._migrate_dead_letters(self.legacy_ip_tag_retry_dead_letter_key) + if migrated: + mythic_sync_log.info("Migrated %s legacy pending or dead-lettered tag update(s)", migrated) + return migrated + + @staticmethod + def _get_query_context(query: DocumentNode, variable_values: dict = None) -> tuple[str, str]: + """Return a GraphQL operation name and serialized representation of its variables.""" + operation_name = "unnamed operation" + for definition in query.definitions: + if definition.name is not None: + operation_name = definition.name.value + break + + diagnostic_variables = { + key: value if key in MythicSync.diagnostic_variable_names else "" + for key, value in (variable_values or {}).items() + } + variables = json.dumps(diagnostic_variables, default=str, sort_keys=True) + return operation_name, variables + + @staticmethod + def _get_transport_error_code(exc: TransportQueryError) -> str | None: + """Return the first GraphQL extension code attached to a transport error.""" + for error in exc.errors or []: + if isinstance(error, dict): + code = error.get("extensions", {}).get("code") + if code: + return code + return None + + async def _wait_for_query_retry(self, operation_name: str, retry_delay: float) -> float: + """Sleep with jitter before a retry and return the next capped base delay.""" + sleep_for = min( + self.max_retry_timeout, + random.uniform(retry_delay * 0.8, retry_delay * 1.2), + ) + mythic_sync_log.warning( + "Retrying Ghostwriter GraphQL operation '%s' in %.1f seconds", + operation_name, + sleep_for, + ) + await asyncio.sleep(sleep_for) + return min(self.max_retry_timeout, retry_delay * 2) + + async def _execute_query( + self, + query: DocumentNode, + variable_values: dict = None, + retry: bool = True, + retry_model_not_found: bool = True, + ) -> dict: """ Execute a GraphQL query against the Ghostwriter server. @@ -274,63 +490,97 @@ async def _execute_query(self, query: DocumentNode, variable_values: dict = None ``variable_values`` The parameters to pass to the query """ + operation_name, variables = self._get_query_context(query, variable_values) + retry_delay = self.wait_timeout while True: try: - try: - result = await self.session.execute(query, variable_values=variable_values) - mythic_sync_log.debug("Successfully executed query with result: %s", result) - return result - except TimeoutError: - mythic_sync_log.error( - "Timeout occurred while trying to connect to Ghostwriter at %s", - self.GHOSTWRITER_URL + result = await self.session.execute(query, variable_values=variable_values) + mythic_sync_log.debug("Successfully executed query with result: %s", result) + return result + except (TimeoutError, builtins.TimeoutError): + mythic_sync_log.error( + "Ghostwriter GraphQL operation '%s' timed out at %s with variables %s", + operation_name, + self.GHOSTWRITER_URL, + variables, + ) + await self._post_error_notification( + f"MythicSync:\nGhostwriter GraphQL operation '{operation_name}' timed out at " + f"{self.GHOSTWRITER_URL} with variables {variables}", + source=f"mythic_sync_query_{operation_name}", + ) + if not retry: + raise + except TransportQueryError as exc: + mythic_sync_log.error( + "Ghostwriter GraphQL operation '%s' failed with variables %s: %s", + operation_name, + variables, + exc, + ) + code = self._get_transport_error_code(exc) + if code == "access-denied": + message = ( + f"Access denied for Ghostwriter GraphQL operation '{operation_name}' with variables " + f"{variables}. Check that the provided service token is valid and has the required " + "permissions." ) - await self._post_error_notification( - f"MythicSync:\nTimeout occurred while trying to connect to Ghostwriter at {self.GHOSTWRITER_URL}", ) - await asyncio.sleep(self.wait_timeout) - continue - except TransportQueryError as e: - mythic_sync_log.exception("Error encountered while fetching GraphQL schema: %s", e) - - payload = e.errors[0] - if "extensions" in payload: - if "code" in payload["extensions"]: - if payload["extensions"]["code"] == "access-denied": - mythic_sync_log.error( - "Access denied for the provided Ghostwriter API token! Check if it is valid, update your configuration, and restart") - await self._post_error_notification( - message=f"Access denied for the provided Ghostwriter API token! Check if it is valid, update your Mythic Sync configuration, and restart the service.", - source="mythic_sync_access_denied", - ) - await asyncio.sleep(self.wait_timeout) - continue - if payload["extensions"]["code"] == "postgres-error": - mythic_sync_log.error( - "Ghostwriter's database rejected the query! Check if your configured log ID is correct.") - await self._post_error_notification( - message=f"Ghostwriter's database rejected the query! Check if your configured log ID ({self.GHOSTWRITER_OPLOG_ID}) is correct.", - source="mythic_sync_reject", - ) - await asyncio.sleep(self.wait_timeout) - continue - await self._post_error_notification( - f"MythicSync:\nError encountered while fetching GraphQL schema: {e}") - await asyncio.sleep(self.wait_timeout) - continue - except GraphQLError as e: - mythic_sync_log.exception("Error with GraphQL query: %s", e) - await self._post_error_notification(f"MythicSync:\nError with GraphQL query: {e}") - await asyncio.sleep(self.wait_timeout) - continue + source = "mythic_sync_access_denied" + retry_delay = max(retry_delay, 60) + elif code == "postgres-error": + message = ( + f"Ghostwriter's database rejected GraphQL operation '{operation_name}' with variables " + f"{variables}. Check if your configured log ID ({self.GHOSTWRITER_OPLOG_ID}) is correct." + ) + source = "mythic_sync_reject" + elif code == "ModelDoesNotExist": + message = ( + "Ghostwriter could not find or authorize the model requested by GraphQL operation " + f"'{operation_name}' with variables {variables}. The referenced oplog entry may not " + "exist, or the service token may not have permission to access it." + ) + source = "mythic_sync_model_not_found" + retry_delay = max(retry_delay, 60) + else: + message = ( + f"MythicSync:\nGhostwriter GraphQL operation '{operation_name}' failed with variables " + f"{variables}: {exc}" + ) + source = f"mythic_sync_query_{operation_name}" + await self._post_error_notification(message=message, source=source) + if not retry or (code == "ModelDoesNotExist" and not retry_model_not_found): + raise + except GraphQLError as exc: + mythic_sync_log.exception( + "Ghostwriter GraphQL operation '%s' failed with variables %s: %s", + operation_name, + variables, + exc, + ) + await self._post_error_notification( + message=f"MythicSync:\nGhostwriter GraphQL operation '{operation_name}' failed with " + f"variables {variables}: {exc}", + source=f"mythic_sync_query_{operation_name}", + ) + if not retry: + raise + except asyncio.CancelledError: + raise except Exception as exc: - mythic_sync_log.error( - "Exception occurred while trying to post the query to Ghostwriter! Trying again in %s seconds...", - self.wait_timeout + mythic_sync_log.exception( + "Unexpected failure in Ghostwriter GraphQL operation '%s' with variables %s", + operation_name, + variables, ) await self._post_error_notification( - f"MythicSync:\nException occurred while trying to post the query to Ghostwriter!\n{exc}") - await asyncio.sleep(self.wait_timeout) - continue + message=f"MythicSync:\nUnexpected failure in Ghostwriter GraphQL operation " + f"'{operation_name}' with variables {variables}: {exc}", + source=f"mythic_sync_query_{operation_name}", + ) + if not retry: + raise + + retry_delay = await self._wait_for_query_retry(operation_name, retry_delay) async def _check_token(self) -> None: """Send a `whoami` query to Ghostwriter to check authentication and token expiration.""" @@ -341,7 +591,11 @@ async def _check_token(self) -> None: if whoami["whoami"]["expires"] == "Never": expiry = "Never" else: - expiry = datetime.fromisoformat(whoami["whoami"]["expires"]) + expiry_value = whoami["whoami"]["expires"] + expiry_value = expiry_value[:-1] + "+00:00" if expiry_value.endswith("Z") else expiry_value + expiry = datetime.fromisoformat(expiry_value) + if expiry.tzinfo is None: + expiry = expiry.replace(tzinfo=timezone.utc) if expiry - now < timedelta(hours=24): mythic_sync_log.debug(f"The provided Ghostwriter API token expires in less than 24 hours ({expiry})!") await self._post_error_notification( @@ -356,19 +610,36 @@ async def _check_token(self) -> None: ) async def _create_initial_entry(self) -> None: - """Send the initial log entry to Ghostwriter's Oplog.""" + """Send one idempotent initialization entry to Ghostwriter's oplog.""" mythic_sync_log.info("Sending the initial Ghostwriter log entry") - variable_values = { - "oplogId": self.GHOSTWRITER_OPLOG_ID, - "description": f"Initial entry from mythic_sync at: {self.MYTHIC_IP}. If you're seeing this then oplog " - f"syncing is working for this C2 server!", - "server": f"Mythic Server ({self.MYTHIC_IP})", - "extraFields": {} - } - await self._execute_query(self.initial_query, variable_values) + entry_identifier = f"mythic_sync:initial:{self.MYTHIC_IP}:{self.MYTHIC_PORT}" + existing_entry = await self._execute_query( + self.entry_identifier_query, + { + "oplog": self.GHOSTWRITER_OPLOG_ID, + "entry_identifier": entry_identifier, + }, + ) + if existing_entry.get("oplogEntry", []): + mythic_sync_log.info( + "Ghostwriter initialization entry already exists for Mythic server %s:%s", + self.MYTHIC_IP, + self.MYTHIC_PORT, + ) + else: + variable_values = { + "oplogId": self.GHOSTWRITER_OPLOG_ID, + "description": f"Initial entry from mythic_sync at: {self.MYTHIC_IP}:{self.MYTHIC_PORT}. " + "If you're seeing this then " + "oplog syncing is working for this C2 server!", + "server": f"Mythic Server ({self.MYTHIC_IP}:{self.MYTHIC_PORT})", + "entry_identifier": entry_identifier, + "extraFields": {}, + } + await self._execute_query(self.initial_query, variable_values) await mythic.send_event_log_message( mythic=self.mythic_instance, - message="Mythic Sync successfully posted its initial log entry to Ghostwriter", + message="Mythic Sync successfully confirmed its initialization entry in Ghostwriter", source="mythic_sync", level="info" ) @@ -376,15 +647,41 @@ async def _create_initial_entry(self) -> None: async def _post_error_notification(self, message: str = None, source: str = None) -> None: """Send an error notification to Mythic's notification center.""" + if self.mythic_instance is None: + mythic_sync_log.debug( + "Skipping Mythic error notification because Mythic authentication is not established" + ) + return if message is None: message = "Mythic Sync logged an error and may need attention to continue syncing.\n" \ "Run this command to review the issue:\n\n" \ " sudo ./mythic-cli logs mythic_sync" + notification_source = "mythic_sync" if source is None else source + now = datetime.now(timezone.utc) + last_notification = self._notification_timestamps.get(notification_source) + if last_notification is not None and now - last_notification < self.last_error_delta: + mythic_sync_log.debug( + "Suppressing duplicate Mythic notification from '%s': %s", + notification_source, + message, + ) + return + mythic_sync_log.info("Submitting an error notification to Mythic's notification center: %s", message) - await mythic.send_event_log_message(mythic=self.mythic_instance, - message=message, - source="mythic_sync" if source is None else source, - level="warning") + try: + await mythic.send_event_log_message(mythic=self.mythic_instance, + message=message, + source=notification_source, + level="warning") + except asyncio.CancelledError: + raise + except Exception: + mythic_sync_log.exception( + "Failed to submit Mythic notification from '%s'", + notification_source, + ) + return + self._notification_timestamps[notification_source] = now return async def _mythic_task_to_ghostwriter_message(self, message: dict) -> dict: @@ -396,37 +693,29 @@ async def _mythic_task_to_ghostwriter_message(self, message: dict) -> dict: ``message`` The message dictionary to be converted """ - gw_message = {} - try: - if message["status_timestamp_submitted"] is not None: - start_date = datetime.strptime( - message["status_timestamp_submitted"], "%Y-%m-%dT%H:%M:%S.%f") - gw_message["startDate"] = start_date.strftime("%Y-%m-%d %H:%M:%S") - if message["status_timestamp_processed"] is not None: - end_date = datetime.strptime( - message["status_timestamp_processed"], "%Y-%m-%dT%H:%M:%S.%f") - gw_message["endDate"] = end_date.strftime("%Y-%m-%d %H:%M:%S") - if message['command'] is not None: - gw_message["command"] = f"{message['command']['cmd']} {message['original_params']}" - else: - gw_message["command"] = f"{message['command_name']} {message['original_params']}" - gw_message["comments"] = message["comment"] if message["comment"] is not None else "" - gw_message["operatorName"] = message["operator"]["username"] if message["operator"] is not None else "" - gw_message["oplog"] = self.GHOSTWRITER_OPLOG_ID - hostname = message["callback"]["host"] - source_ip = await self._get_sorted_ips(message["callback"]["ip"]) - gw_message["sourceIp"] = f"{hostname} ({source_ip})" - gw_message[ - "description"] = f"PID: {message['callback']['pid']}, Callback: {message['callback']['display_id']}" - gw_message["userContext"] = message["callback"]["user"] - gw_message["tool"] = message["callback"]["payload"]["payloadtype"]["name"] - gw_message['entry_identifier'] = message["agent_task_id"] - gw_message['tags'] = [f"mythic:{x['tagtype']['name']}" for x in message['tags']] - gw_message['extraFields'] = {} - except Exception: - mythic_sync_log.exception( - "Encountered an exception while processing Mythic's message into a message for Ghostwriter" - ) + callback = message["callback"] + hostname = callback["host"] + source_ip = await self._get_sorted_ips(callback.get("ip")) + command = message.get("command") + command_name = command["cmd"] if command is not None else message["command_name"] + + gw_message = { + "command": f"{command_name} {message.get('original_params', '')}".rstrip(), + "comments": message.get("comment") or "", + "operatorName": (message.get("operator") or {}).get("username", ""), + "oplog": self.GHOSTWRITER_OPLOG_ID, + "sourceIp": f"{hostname} ({source_ip})" if source_ip else hostname, + "description": f"PID: {callback['pid']}, Callback: {callback['display_id']}", + "userContext": callback["user"], + "tool": callback["payload"]["payloadtype"]["name"], + "entry_identifier": message["agent_task_id"], + "tags": [f"mythic:{tag['tagtype']['name']}" for tag in message.get("tags", [])], + "extraFields": {}, + } + if message.get("status_timestamp_submitted"): + gw_message["startDate"] = self._parse_mythic_timestamp(message["status_timestamp_submitted"]) + if message.get("status_timestamp_processed"): + gw_message["endDate"] = self._parse_mythic_timestamp(message["status_timestamp_processed"]) return gw_message async def _mythic_callback_to_ghostwriter_message(self, message: dict) -> dict: @@ -438,54 +727,283 @@ async def _mythic_callback_to_ghostwriter_message(self, message: dict) -> dict: ``message`` The message dictionary to be converted """ - gw_message = {} + source_ip = await self._get_sorted_ips(message.get("ip")) + integrity_level = message.get("integrity_level") + integrity = self.integrity_levels.get(integrity_level, f"Unknown ({integrity_level})") + opsys = (message.get("os") or "").replace("\n", ", ") + hostname = message["host"] + return { + "startDate": self._parse_mythic_timestamp(message["init_callback"]), + "comments": f"New Callback {message['display_id']}", + "description": ( + f"Computer: {hostname}, Integrity Level: {integrity}, Process: {message['process_name']}, " + f"PID: {message['pid']}, User: {message['user']}, Domain: {message['domain']}, OS: {opsys}" + ), + "operatorName": (message.get("operator") or {}).get("username", ""), + "sourceIp": f"{hostname} ({source_ip})" if source_ip else hostname, + "userContext": message["user"], + "tool": message["payload"]["payloadtype"]["name"], + "oplog": self.GHOSTWRITER_OPLOG_ID, + "entry_identifier": message["agent_callback_id"], + "extraFields": {}, + "command": "", + } + + def _queue_task_tags(self, tags: list, entry_id: str, entry_identifier: str) -> None: + """Persist the latest desired Mythic tags for asynchronous processing.""" + entry_id = str(entry_id) + payload = json.dumps( + { + "attempt": 0, + "entry_id": entry_id, + "entry_identifier": entry_identifier, + "tags": tags, + }, + sort_keys=True, + ) + pipeline = self.rconn.pipeline(transaction=True) + pipeline.hset(self.tag_retry_data_key, entry_id, payload) + pipeline.zadd(self.tag_retry_schedule_key, {entry_id: time.time()}) + pipeline.execute() + + async def _handle_task_tags(self, tags: list, entry_id: str, retry: bool = False) -> None: + """Merge Mythic tags into one Ghostwriter oplog entry.""" + current_tags = await self._execute_query( + self.query_oplog_entry_tags, + {"oplog_entry_id": entry_id}, + retry=retry, + ) + updated_tags = [] + for current_tag in current_tags['tags']['tags']: + if current_tag.startswith("mythic:"): + if current_tag in tags: + updated_tags.append(current_tag) + else: + updated_tags.append(current_tag) + for current_tag in tags: + if current_tag not in updated_tags: + updated_tags.append(current_tag) + await self._execute_query( + self.set_tags_mutation, + {"oplog_entry_id": entry_id, "tags": updated_tags}, + retry=retry, + ) + + def _remove_pending_tag_update(self, entry_id: str, expected_payload: bytes | str) -> bool: + """Remove a pending job only when it has not been replaced by newer tag data.""" + if self.rconn.hget(self.tag_retry_data_key, entry_id) != expected_payload: + return False + pipeline = self.rconn.pipeline(transaction=True) + pipeline.hdel(self.tag_retry_data_key, entry_id) + pipeline.zrem(self.tag_retry_schedule_key, entry_id) + pipeline.execute() + return True + + def _dead_letter_pending_tag_update( + self, + entry_id: str, + payload: dict, + expected_payload: bytes | str, + reason: str, + ) -> bool: + """Retain an irrecoverable tag job for diagnosis while removing it from active retries.""" + if self.rconn.hget(self.tag_retry_data_key, entry_id) != expected_payload: + return False + dead_letter = dict(payload) + dead_letter["dead_lettered_at"] = datetime.now(timezone.utc).isoformat() + dead_letter["reason"] = reason + pipeline = self.rconn.pipeline(transaction=True) + pipeline.hdel(self.tag_retry_data_key, entry_id) + pipeline.zrem(self.tag_retry_schedule_key, entry_id) + pipeline.hset( + self.tag_retry_dead_letter_key, + entry_id, + json.dumps(dead_letter, sort_keys=True), + ) + pipeline.execute() + return True + + def _reschedule_pending_tag_update( + self, + entry_id: str, + payload: dict, + expected_payload: bytes | str, + ) -> None: + """Apply capped exponential backoff to a pending tag job.""" + if self.rconn.hget(self.tag_retry_data_key, entry_id) != expected_payload: + return + attempt = payload.get("attempt", 0) + 1 + retry_delay = min( + self.max_retry_timeout, + self.wait_timeout * (2 ** min(attempt, 10)), + ) + retry_delay = min( + self.max_retry_timeout, + random.uniform(retry_delay * 0.8, retry_delay * 1.2), + ) + payload["attempt"] = attempt + pipeline = self.rconn.pipeline(transaction=True) + pipeline.hset( + self.tag_retry_data_key, + entry_id, + json.dumps(payload, sort_keys=True), + ) + pipeline.zadd( + self.tag_retry_schedule_key, + {entry_id: time.time() + retry_delay}, + ) + pipeline.execute() + mythic_sync_log.warning( + "Tag update for Ghostwriter oplog entry %s failed; retrying in %.1f seconds", + entry_id, + retry_delay, + ) + + async def _reconcile_pending_tag_update( + self, + stale_entry_id: str, + payload: dict, + expected_payload: bytes | str, + ) -> None: + """Move a stale tag job to the current entry or retire it when the entry was deleted.""" + entry_identifier = payload.get("entry_identifier") + if not entry_identifier: + if self._dead_letter_pending_tag_update( + stale_entry_id, + payload, + expected_payload, + "legacy job has no entry identifier", + ): + mythic_sync_log.error( + "Dead-lettered legacy tag job for missing Ghostwriter entry %s because it has no entry identifier", + stale_entry_id, + ) + await self._post_error_notification(source="mythic_sync_orphaned_tag_update") + return + + query_result = await self._execute_query( + self.entry_identifier_query, + { + "oplog": self.GHOSTWRITER_OPLOG_ID, + "entry_identifier": entry_identifier, + }, + retry=False, + retry_model_not_found=False, + ) + existing_entries = query_result.get("oplogEntry", []) if query_result else [] + if not existing_entries: + if self._dead_letter_pending_tag_update( + stale_entry_id, + payload, + expected_payload, + "entry identifier no longer resolves in Ghostwriter", + ): + mythic_sync_log.error( + "Dead-lettered tag job for deleted or inaccessible Ghostwriter entry %s (entryIdentifier %s)", + stale_entry_id, + entry_identifier, + ) + await self._post_error_notification( + message=( + "Mythic Sync stopped retrying a tag update for Ghostwriter entry " + f"{stale_entry_id} (entryIdentifier {entry_identifier}) because the entry no longer " + "resolves. The job remains in the Redis dead-letter hash for diagnosis." + ), + source="mythic_sync_orphaned_tag_update", + ) + return + + current_entry_id = str(existing_entries[0]["id"]) + if current_entry_id == stale_entry_id: + raise RuntimeError( + "Ghostwriter resolved entryIdentifier %s to entry %s, but its tag API reported that entry missing" + % (entry_identifier, stale_entry_id) + ) + if self.rconn.hget(self.tag_retry_data_key, stale_entry_id) != expected_payload: + return + + payload["attempt"] = 0 + payload["entry_id"] = current_entry_id + pipeline = self.rconn.pipeline(transaction=True) + pipeline.hdel(self.tag_retry_data_key, stale_entry_id) + pipeline.zrem(self.tag_retry_schedule_key, stale_entry_id) + pipeline.hset(self.tag_retry_data_key, current_entry_id, json.dumps(payload, sort_keys=True)) + pipeline.zadd(self.tag_retry_schedule_key, {current_entry_id: time.time()}) + pipeline.execute() + self._set_cached_entry_id(entry_identifier, current_entry_id) + mythic_sync_log.info( + "Moved pending tag update from stale Ghostwriter entry %s to entry %s", + stale_entry_id, + current_entry_id, + ) + + async def _process_pending_tag_update(self, entry_id: str) -> None: + """Attempt one queued tag update and reschedule it on any failure.""" + payload_raw = self.rconn.hget(self.tag_retry_data_key, entry_id) + if payload_raw is None: + self.rconn.zrem(self.tag_retry_schedule_key, entry_id) + return + + payload = json.loads(payload_raw) try: - callback_date = datetime.strptime(message["init_callback"], "%Y-%m-%dT%H:%M:%S.%f") - gw_message["startDate"] = callback_date.strftime("%Y-%m-%d %H:%M:%S") - gw_message["comments"] = f"New Callback {message['display_id']}" - integrity = self.integrity_levels[message["integrity_level"]] - opsys = message['os'].replace("\n", ", ") - gw_message[ - "description"] = f"Computer: {message['host']}, Integrity Level: {integrity}, Process: {message['process_name']}, PID: {message['pid']}, User: {message['user']}, Domain: {message['domain']}, OS: {opsys}" - gw_message["operatorName"] = message["operator"]["username"] if message["operator"] is not None else "" - source_ip = await self._get_sorted_ips(message["ip"]) - gw_message["sourceIp"] = f"{message['host']} ({source_ip})" - gw_message["userContext"] = message["user"] - gw_message["tool"] = message["payload"]["payloadtype"]["name"] - gw_message["oplog"] = self.GHOSTWRITER_OPLOG_ID - gw_message['entry_identifier'] = message["agent_callback_id"] - gw_message['extraFields'] = {} - gw_message["command"] = "" + await self._handle_task_tags(payload["tags"], payload["entry_id"], retry=False) + except TransportQueryError as exc: + if self._get_transport_error_code(exc) == "ModelDoesNotExist": + try: + await self._reconcile_pending_tag_update(entry_id, payload, payload_raw) + return + except asyncio.CancelledError: + raise + except Exception: + mythic_sync_log.exception( + "Failed to reconcile missing Ghostwriter entry %s for a pending tag update", + entry_id, + ) + self._reschedule_pending_tag_update(entry_id, payload, payload_raw) + return + except asyncio.CancelledError: + raise except Exception: - mythic_sync_log.exception( - "Encountered an exception while processing Mythic's message into a message for Ghostwriter! Received message: %s", - message - ) - return gw_message + self._reschedule_pending_tag_update(entry_id, payload, payload_raw) + return + + self._remove_pending_tag_update(entry_id, payload_raw) + + async def retry_pending_tags(self) -> None: + """Continuously process due tag updates without blocking log ingestion.""" + mythic_sync_log.info("Starting pending Ghostwriter tag update worker") + while True: + try: + due_entries = self.rconn.zrangebyscore( + self.tag_retry_schedule_key, + 0, + time.time(), + start=0, + num=1, + ) + if due_entries: + entry_id = due_entries[0] + if isinstance(entry_id, bytes): + entry_id = entry_id.decode() + await self._process_pending_tag_update(str(entry_id)) + continue + except asyncio.CancelledError: + raise + except Exception: + mythic_sync_log.exception("Failed while processing pending Ghostwriter tag updates") + await self._post_error_notification(source="mythic_sync_tag_retry_worker") + await asyncio.sleep(self.tag_retry_poll_interval) + + @staticmethod + def _get_returning_entry_id(result: dict, mutation_name: str) -> str: + """Return the first entry ID from a Ghostwriter mutation response, if present.""" + if not result or mutation_name not in result: + return "" + returning = result[mutation_name].get("returning", []) + if not returning: + return "" + return str(returning[0]["id"]) - async def _handleTaskTags(self, tags: list, entry_id: str) -> None: - """ - Process a Mythic Task's tags for an oplog entry - :param message: - :return: - """ - # given an oplog_entry id, fetch its current tags, merge it with message['tags'] - currentTags = await self._execute_query(self.query_oplog_entry_tags, { - "oplog_entry_id": entry_id, - }) - updatedTags = [] - for currentTag in currentTags['tags']['tags']: - if currentTag.startswith("mythic:"): - if currentTag in tags: - updatedTags.append(currentTag) - else: - updatedTags.append(currentTag) - for currentTag in tags: - updatedTags.append(currentTag) - await self._execute_query(self.set_tags_mutation, { - "oplog_entry_id": entry_id, - "tags": updatedTags - }) async def _create_entry(self, message: dict) -> None: """ Create an entry for a Mythic task in Ghostwriter's ``OplogEntry`` model. Uses the @@ -498,19 +1016,26 @@ async def _create_entry(self, message: dict) -> None: """ entry_id = "" gw_message = {} - if "agent_task_id" in message: - entry_id = message["agent_task_id"] - mythic_sync_log.debug(f"Adding task: {message['agent_task_id']}") - gw_message = await self._mythic_task_to_ghostwriter_message(message) - elif "agent_callback_id" in message: - entry_id = message["agent_callback_id"] - mythic_sync_log.debug(f"Adding callback: {message['agent_callback_id']}") - gw_message = await self._mythic_callback_to_ghostwriter_message(message) - else: - mythic_sync_log.error( - "Failed to create an entry for task, no `agent_task_id` or `agent_callback_id` found! Message " - "contents: %s", message - ) + try: + if "agent_task_id" in message: + entry_id = message["agent_task_id"] + mythic_sync_log.debug(f"Adding task: {message['agent_task_id']}") + gw_message = await self._mythic_task_to_ghostwriter_message(message) + elif "agent_callback_id" in message: + entry_id = message["agent_callback_id"] + mythic_sync_log.debug(f"Adding callback: {message['agent_callback_id']}") + gw_message = await self._mythic_callback_to_ghostwriter_message(message) + else: + raise ValueError( + "Message has no `agent_task_id` or `agent_callback_id`; received keys: " + f"{sorted(message)}" + ) + except asyncio.CancelledError: + raise + except Exception: + mythic_sync_log.exception("Failed to convert Mythic event %s for Ghostwriter", entry_id or "unknown") + await self._post_error_notification(source="mythic_sync_conversion") + raise tags = gw_message.pop("tags", []) if entry_id != "" and 'entry_identifier' in gw_message: result = None @@ -520,28 +1045,33 @@ async def _create_entry(self, message: dict) -> None: "entry_identifier": gw_message['entry_identifier'], }) if query_result and "oplogEntry" in query_result and len(query_result["oplogEntry"]) > 0: + ghostwriter_entry_id = str(query_result["oplogEntry"][0]["id"]) mythic_sync_log.info( f"Duplicate entry found based on entryIdentifier, {gw_message['entry_identifier']}, not sending") # save off id of oplog entry with this gw_message['entry_identifier'] so we don't try to send it again - self.rconn.set(entry_id, query_result["oplogEntry"][0]["id"]) - await self._handleTaskTags(tags, query_result["oplogEntry"][0]["id"]) + self._set_cached_entry_id(entry_id, ghostwriter_entry_id) + self._queue_task_tags(tags, ghostwriter_entry_id, gw_message["entry_identifier"]) return result = await self._execute_query(self.insert_query, gw_message) - if result and "insert_oplogEntry" in result: + ghostwriter_entry_id = self._get_returning_entry_id(result, "insert_oplogEntry") + if ghostwriter_entry_id: # JSON response example: `{'data': {'insert_oplogEntry': {'returning': [{'id': 192}]}}}` - self.rconn.set(entry_id, result["insert_oplogEntry"]["returning"][0]["id"]) - await self._handleTaskTags(tags, result["insert_oplogEntry"]["returning"][0]["id"]) + self._set_cached_entry_id(entry_id, ghostwriter_entry_id) + self._queue_task_tags(tags, ghostwriter_entry_id, gw_message["entry_identifier"]) else: - mythic_sync_log.info( - "Did not receive a response with data from Ghostwriter's GraphQL API! Response: %s", + raise RuntimeError( + "Ghostwriter did not return an inserted oplog entry ID. Response: %s" % result ) + except asyncio.CancelledError: + raise except Exception: mythic_sync_log.exception( "Encountered an exception while trying to create a new log entry! Response from Ghostwriter: %s", result, ) await self._post_error_notification() + raise async def _update_entry(self, message: dict, entry_id: str) -> None: """ @@ -556,20 +1086,83 @@ async def _update_entry(self, message: dict, entry_id: str) -> None: The ID of the log entry to be updated """ mythic_sync_log.debug(f"Updating task: {message['agent_task_id']} - {message['id']} : {entry_id}") - gw_message = await self._mythic_task_to_ghostwriter_message(message) + try: + gw_message = await self._mythic_task_to_ghostwriter_message(message) + except asyncio.CancelledError: + raise + except Exception: + mythic_sync_log.exception( + "Failed to convert Mythic task %s for a Ghostwriter update", + message["agent_task_id"], + ) + await self._post_error_notification(source="mythic_sync_conversion") + raise gw_message["id"] = entry_id tags = gw_message.pop("tags", []) try: - result = await self._execute_query(self.update_query, gw_message) - if not result or "update_oplogEntry" not in result: - mythic_sync_log.info( - "Did not receive a response with data from Ghostwriter's GraphQL API! Response: %s", - result + try: + result = await self._execute_query( + self.update_query, + gw_message, + retry_model_not_found=False, ) - else: - await self._handleTaskTags(tags, entry_id) + except TransportQueryError as exc: + if self._get_transport_error_code(exc) != "ModelDoesNotExist": + raise + result = {} + ghostwriter_entry_id = self._get_returning_entry_id(result, "update_oplogEntry") + if not ghostwriter_entry_id: + mythic_entry_id = message["agent_task_id"] + stale_entry_id = gw_message.pop("id") + mythic_sync_log.warning( + "Ghostwriter oplog entry %s cached for Mythic task %s no longer exists or is inaccessible; " + "reconciling with entryIdentifier %s", + stale_entry_id, + mythic_entry_id, + gw_message["entry_identifier"], + ) + self._delete_cached_entry_id(mythic_entry_id) + + query_result = await self._execute_query( + self.entry_identifier_query, + { + "oplog": gw_message["oplog"], + "entry_identifier": gw_message["entry_identifier"], + }, + ) + existing_entries = query_result.get("oplogEntry", []) if query_result else [] + if existing_entries: + ghostwriter_entry_id = str(existing_entries[0]["id"]) + gw_message["id"] = ghostwriter_entry_id + retry_result = await self._execute_query(self.update_query, gw_message) + updated_entry_id = self._get_returning_entry_id(retry_result, "update_oplogEntry") + if not updated_entry_id: + raise RuntimeError( + "Ghostwriter found oplog entry %s by entryIdentifier but did not update it. " + "Response: %s" % (ghostwriter_entry_id, retry_result) + ) + ghostwriter_entry_id = updated_entry_id + else: + insert_result = await self._execute_query(self.insert_query, gw_message) + ghostwriter_entry_id = self._get_returning_entry_id( + insert_result, + "insert_oplogEntry", + ) + if not ghostwriter_entry_id: + raise RuntimeError( + "Ghostwriter did not return an ID while recreating stale oplog entry %s. " + "Response: %s" % (stale_entry_id, insert_result) + ) + + self._set_cached_entry_id(mythic_entry_id, ghostwriter_entry_id) + + self._queue_task_tags(tags, ghostwriter_entry_id, gw_message["entry_identifier"]) + except asyncio.CancelledError: + raise except Exception: mythic_sync_log.exception("Exception encountered while trying to update task log entry in Ghostwriter!") + await self._post_error_notification(source="mythic_sync_update_entry") + raise async def handle_task(self) -> None: """ @@ -615,14 +1208,16 @@ async def handle_task(self) -> None: mythic=self.mythic_instance, custom_return_attributes=custom_return_attributes, ): try: - entry_id = self.rconn.get(data["agent_task_id"]) + entry_id = self._get_cached_entry_id(data["agent_task_id"]) + except asyncio.CancelledError: + raise except Exception: mythic_sync_log.exception( - "Encountered an exception while connecting to Redis to fetch data! Data returned by Mythic: %s", - data + "Encountered an exception while fetching the Redis mapping for Mythic task %s", + data.get("agent_task_id", "unknown"), ) await self._post_error_notification() - continue + raise if entry_id is not None: await self._update_entry(data, entry_id.decode()) else: @@ -677,6 +1272,8 @@ async def _wait_for_service(self) -> None: ) await asyncio.sleep(self.wait_timeout) continue + except asyncio.CancelledError: + raise except Exception as e: await asyncio.sleep(self.wait_timeout) mythic_sync_log.warning("failed to connect to Mythic: %s", e) @@ -684,15 +1281,55 @@ async def _wait_for_service(self) -> None: return async def _wait_for_redis(self) -> None: - """Wait for a connection to be established with Mythic's Redis container.""" + """Wait for Redis to accept commands and report the selected durable state.""" while True: try: - self.rconn = redis.Redis(host=self.REDIS_HOSTNAME, port=self.REDIS_PORT, db=1) + if self.REDIS_URL: + self.rconn = redis.Redis.from_url( + self.REDIS_URL, + socket_connect_timeout=5, + socket_timeout=5, + ) + redis_target = "the configured REDIS_URL" + else: + self.rconn = redis.Redis( + host=self.REDIS_HOSTNAME, + port=self.REDIS_PORT, + db=self.REDIS_DB, + socket_connect_timeout=5, + socket_timeout=5, + ) + redis_target = f"{self.REDIS_HOSTNAME}:{self.REDIS_PORT} database {self.REDIS_DB}" + self.rconn.ping() + self._migrate_legacy_tag_queue() + pending_tags = self.rconn.hlen(self.tag_retry_data_key) + dead_letter_tags = self.rconn.hlen(self.tag_retry_dead_letter_key) + mythic_sync_log.info( + "Connected to Redis at %s; %s tag update(s) pending, %s dead-lettered", + redis_target, + pending_tags, + dead_letter_tags, + ) + if not self.REDIS_URL and self.REDIS_HOSTNAME.lower() in {"127.0.0.1", "localhost"}: + mythic_sync_log.warning( + "Using embedded Redis; mount stable persistent storage at /data before relying on " + "queued jobs to survive Mythic service recreation" + ) + if dead_letter_tags: + mythic_sync_log.warning( + "%s tag update(s) require operator review in Redis hash '%s'; inspect with HGETALL", + dead_letter_tags, + self.tag_retry_dead_letter_key, + ) return + except asyncio.CancelledError: + raise except Exception: mythic_sync_log.exception( - "Encountered an exception while trying to connect to Redis, %s:%s, trying again in %s seconds...", - self.REDIS_HOSTNAME, self.REDIS_PORT, self.wait_timeout + "Encountered an exception while trying to connect to Redis at %s, trying again in %s seconds...", + "the configured REDIS_URL" if self.REDIS_URL else + f"{self.REDIS_HOSTNAME}:{self.REDIS_PORT} database {self.REDIS_DB}", + self.wait_timeout, ) await self._post_error_notification() await asyncio.sleep(self.wait_timeout) @@ -701,54 +1338,67 @@ async def _wait_for_redis(self) -> None: async def __wait_for_authentication(self) -> mythic_classes.Mythic: """Wait for authentication with Mythic to complete.""" while True: - # If ``MYTHIC_API_KEY`` is not set in the environment, then authenticate with user credentials - if len(self.MYTHIC_API_KEY) == 0: + if self.MYTHIC_API_KEY: mythic_sync_log.info( - "Authenticating to Mythic, https://%s:%s, with username and password", - self.MYTHIC_IP, self.MYTHIC_PORT + "Authenticating to Mythic, https://%s:%s, with a specified API key", + self.MYTHIC_IP, + self.MYTHIC_PORT, ) try: mythic_instance = await mythic.login( - username=self.MYTHIC_USERNAME, - password=self.MYTHIC_PASSWORD, + apitoken=self.MYTHIC_API_KEY, server_ip=self.MYTHIC_IP, server_port=self.MYTHIC_PORT, ssl=True, - timeout=-1) - except Exception as e: - mythic_sync_log.error( - "Encountered an exception while trying to authenticate to Mythic, trying again in %s seconds...", - self.wait_timeout ) - await asyncio.sleep(self.wait_timeout) - continue - try: await mythic.get_me(mythic=mythic_instance) - except Exception as e: + except asyncio.CancelledError: + raise + except Exception as exc: mythic_sync_log.error( - "Encountered an exception while trying to get user info from Mythic, trying again in %s seconds...", - self.wait_timeout + "Failed to authenticate with the Mythic API key: %s; trying again in %s seconds...", + exc, + self.wait_timeout, ) await asyncio.sleep(self.wait_timeout) continue - elif self.MYTHIC_USERNAME == "" and self.MYTHIC_PASSWORD == "": - mythic_sync_log.error("You must supply a MYTHIC_USERNAME and MYTHIC_PASSWORD") - sys.exit(1) else: + if not self.MYTHIC_USERNAME or not self.MYTHIC_PASSWORD: + raise RuntimeError( + "MYTHIC_API_KEY or both MYTHIC_USERNAME and MYTHIC_PASSWORD must be supplied" + ) mythic_sync_log.info( - "Authenticating to Mythic, https://%s:%s, with a specified API Key", + "Authenticating to Mythic, https://%s:%s, with username and password", self.MYTHIC_IP, self.MYTHIC_PORT ) try: mythic_instance = await mythic.login( - apitoken=self.MYTHIC_API_KEY, + username=self.MYTHIC_USERNAME, + password=self.MYTHIC_PASSWORD, server_ip=self.MYTHIC_IP, server_port=self.MYTHIC_PORT, - ssl=True) + ssl=True, + timeout=-1) + except asyncio.CancelledError: + raise + except Exception as exc: + mythic_sync_log.error( + "Encountered an exception while trying to authenticate to Mythic: %s; trying again in %s " + "seconds...", + exc, + self.wait_timeout + ) + await asyncio.sleep(self.wait_timeout) + continue + try: await mythic.get_me(mythic=mythic_instance) - except Exception as e: + except asyncio.CancelledError: + raise + except Exception as exc: mythic_sync_log.error( - "Failed to authenticate with the Mythic API token, trying again in %s seconds...", + "Encountered an exception while trying to get user info from Mythic: %s; trying again in " + "%s seconds...", + exc, self.wait_timeout ) await asyncio.sleep(self.wait_timeout) @@ -761,17 +1411,24 @@ async def scripting(): mythic_sync = MythicSync() while True: await mythic_sync.initialize() + tasks = [ + asyncio.create_task(mythic_sync.handle_task()), + asyncio.create_task(mythic_sync.handle_callback()), + asyncio.create_task(mythic_sync.retry_pending_tags()), + ] try: - _ = await asyncio.gather( - mythic_sync.handle_task(), - mythic_sync.handle_callback(), - ) + await asyncio.gather(*tasks) + except asyncio.CancelledError: + raise except Exception: mythic_sync_log.exception( "Encountered an exception while subscribing to tasks and responses, restarting..." ) finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) await mythic_sync.client.close_async() - -asyncio.run(scripting()) \ No newline at end of file +if __name__ == "__main__": + asyncio.run(scripting()) diff --git a/test_sync.py b/test_sync.py new file mode 100644 index 0000000..622057d --- /dev/null +++ b/test_sync.py @@ -0,0 +1,528 @@ +import asyncio +import os +import subprocess +import sys +import unittest +from unittest.mock import AsyncMock, patch + +from gql.transport.exceptions import TransportQueryError + + +os.environ.setdefault("MYTHIC_IP", "127.0.0.1") +os.environ.setdefault("MYTHIC_PORT", "7443") +os.environ.setdefault("GHOSTWRITER_API_KEY", "gwst_test") +os.environ.setdefault("GHOSTWRITER_URL", "http://ghostwriter") +os.environ.setdefault("GHOSTWRITER_OPLOG_ID", "12") + +from sync import MythicSync # noqa: E402 + + +def model_not_found_error(): + return TransportQueryError( + "Not Found", + errors=[{"message": "Not Found", "extensions": {"code": "ModelDoesNotExist"}}], + ) + + +class FakeRedis: + def __init__(self): + self.values = {} + self.hashes = {} + self.sorted_sets = {} + + @staticmethod + def _bytes(value): + if value is None or isinstance(value, bytes): + return value + return str(value).encode() + + @staticmethod + def _key(value): + return value.decode() if isinstance(value, bytes) else str(value) + + def set(self, key, value): + self.values[self._key(key)] = self._bytes(value) + + def get(self, key): + return self.values.get(self._key(key)) + + def delete(self, key): + key = self._key(key) + self.values.pop(key, None) + self.hashes.pop(key, None) + self.sorted_sets.pop(key, None) + + def pipeline(self, transaction=True): + return self + + def execute(self): + return [] + + def ping(self): + return True + + def hset(self, name, key, value): + self.hashes.setdefault(name, {})[self._key(key)] = self._bytes(value) + + def hget(self, name, key): + return self.hashes.get(name, {}).get(self._key(key)) + + def hgetall(self, name): + return { + key.encode(): value + for key, value in self.hashes.get(name, {}).items() + } + + def hlen(self, name): + return len(self.hashes.get(name, {})) + + def hdel(self, name, key): + self.hashes.get(name, {}).pop(self._key(key), None) + + def zadd(self, name, mapping): + target = self.sorted_sets.setdefault(name, {}) + for key, score in mapping.items(): + target[self._key(key)] = score + + def zrem(self, name, key): + self.sorted_sets.get(name, {}).pop(self._key(key), None) + + def zrangebyscore(self, name, minimum, maximum, start=0, num=None): + matches = [ + key.encode() + for key, score in sorted(self.sorted_sets.get(name, {}).items(), key=lambda item: item[1]) + if minimum <= score <= maximum + ] + return matches[start:] if num is None else matches[start:start + num] + + def zrange(self, name, start, end, withscores=False): + matches = sorted(self.sorted_sets.get(name, {}).items(), key=lambda item: item[1]) + if end >= 0: + matches = matches[start:end + 1] + else: + matches = matches[start:] + if withscores: + return [(key.encode(), score) for key, score in matches] + return [key.encode() for key, _ in matches] + + +class MythicSyncTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.sync = MythicSync() + self.sync.REDIS_URL = "" + self.sync.rconn = FakeRedis() + self.sync._post_error_notification = AsyncMock() + + def test_redis_namespace_includes_mythic_port(self): + self.assertEqual( + self.sync.redis_namespace, + f"mythic_sync:{self.sync.GHOSTWRITER_OPLOG_ID}:{self.sync.MYTHIC_IP}:{self.sync.MYTHIC_PORT}", + ) + + def test_mythic_port_defaults_to_7443_when_omitted(self): + environment = os.environ.copy() + environment.pop("MYTHIC_PORT", None) + + result = subprocess.run( + [ + sys.executable, + "-c", + "from sync import MythicSync; print(MythicSync.MYTHIC_PORT)", + ], + cwd=os.path.dirname(__file__), + env=environment, + capture_output=True, + text=True, + check=True, + ) + + self.assertEqual(result.stdout.strip(), "7443") + + async def test_get_sorted_ips_returns_plain_sorted_string(self): + result = await self.sync._get_sorted_ips( + '["10.0.0.20/24", "", "invalid", "10.0.0.3", "2001:db8::1"]' + ) + + self.assertEqual(result, "10.0.0.3, 10.0.0.20, 2001:db8::1") + + async def test_get_sorted_ips_accepts_a_single_plain_ip(self): + self.assertEqual(await self.sync._get_sorted_ips("10.0.0.1/24"), "10.0.0.1") + + async def test_get_sorted_ips_omits_low_value_adapter_addresses(self): + result = await self.sync._get_sorted_ips([ + "10.10.14.4", + "100.76.103.80", + "172.17.0.1", + "172.18.0.1", + "192.168.3.150", + "dead:beef:2::1002", + "fd7a:115c:a1e0::1437:6753", + "fe80::1caf:f2ff:fe76:822", + "fe80::598e:5d6:f491:e319", + "127.0.0.1", + "::", + "ff02::1", + ]) + + self.assertEqual( + result, + "10.10.14.4, 100.76.103.80, 172.17.0.1, 172.18.0.1, 192.168.3.150, " + "dead:beef:2::1002, fd7a:115c:a1e0::1437:6753", + ) + + def test_parse_mythic_timestamp_accepts_common_variants(self): + self.assertEqual( + self.sync._parse_mythic_timestamp("2026-07-18T12:34:56Z"), + "2026-07-18T12:34:56+00:00", + ) + self.assertEqual( + self.sync._parse_mythic_timestamp("2026-07-18T12:34:56.123456"), + "2026-07-18T12:34:56.123456", + ) + + def test_query_context_redacts_sensitive_values(self): + operation, variables = self.sync._get_query_context(self.sync.update_query, { + "id": "42", + "oplog": "12", + "command": "secret command", + "comments": "sensitive comment", + }) + + self.assertEqual(operation, "UpdateMythicSyncLog") + self.assertIn('"id": "42"', variables) + self.assertIn('"oplog": "12"', variables) + self.assertNotIn("secret command", variables) + self.assertNotIn("sensitive comment", variables) + + async def test_query_retries_with_backoff_then_resets_on_success(self): + self.sync.session = AsyncMock() + self.sync.session.execute.side_effect = [TimeoutError(), {"whoami": {"expires": "Never"}}] + + with patch("sync.random.uniform", return_value=5), patch( + "sync.asyncio.sleep", new_callable=AsyncMock + ) as sleep: + result = await self.sync._execute_query(self.sync.whoami_query) + + self.assertEqual(result, {"whoami": {"expires": "Never"}}) + sleep.assert_awaited_once_with(5) + self.assertEqual(self.sync.session.execute.await_count, 2) + + async def test_model_not_found_can_exit_retry_loop_for_reconciliation(self): + self.sync.session = AsyncMock() + self.sync.session.execute.side_effect = model_not_found_error() + + with patch("sync.asyncio.sleep", new_callable=AsyncMock) as sleep: + with self.assertRaises(TransportQueryError): + await self.sync._execute_query( + self.sync.update_query, + {"id": "41", "oplog": "12"}, + retry_model_not_found=False, + ) + + self.assertEqual(self.sync.session.execute.await_count, 1) + sleep.assert_not_awaited() + + async def test_token_expiration_accepts_z_timezone(self): + self.sync._execute_query = AsyncMock(return_value={ + "whoami": {"expires": "2099-01-01T00:00:00Z"}, + }) + + with patch("sync.mythic.send_event_log_message", new=AsyncMock()): + await self.sync._check_token() + + async def test_stale_cached_id_is_repaired_from_entry_identifier(self): + self.sync.rconn.set("task-1", "41") + self.sync._mythic_task_to_ghostwriter_message = AsyncMock(return_value={ + "entry_identifier": "task-1", + "oplog": "12", + "tags": ["mythic:test"], + }) + self.sync._execute_query = AsyncMock(side_effect=[ + {"update_oplogEntry": {"returning": []}}, + {"oplogEntry": [{"id": "99"}]}, + {"update_oplogEntry": {"returning": [{"id": "99"}]}}, + ]) + + await self.sync._update_entry({"agent_task_id": "task-1", "id": "mythic-row"}, "41") + + self.assertEqual(self.sync.rconn.get(self.sync._redis_entry_key("task-1")), b"99") + self.assertIsNone(self.sync.rconn.get("task-1")) + self.assertIsNotNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "99")) + + async def test_deleted_entry_is_recreated_from_current_task(self): + self.sync.rconn.set("task-1", "41") + self.sync._mythic_task_to_ghostwriter_message = AsyncMock(return_value={ + "entry_identifier": "task-1", + "oplog": "12", + "tags": [], + }) + self.sync._execute_query = AsyncMock(side_effect=[ + {"update_oplogEntry": {"returning": []}}, + {"oplogEntry": []}, + {"insert_oplogEntry": {"returning": [{"id": "77"}]}}, + ]) + + await self.sync._update_entry({"agent_task_id": "task-1", "id": "mythic-row"}, "41") + + self.assertEqual(self.sync.rconn.get(self.sync._redis_entry_key("task-1")), b"77") + self.assertIsNone(self.sync.rconn.get("task-1")) + self.assertIsNotNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "77")) + + async def test_model_not_found_update_reconciles_instead_of_retrying_forever(self): + self.sync._mythic_task_to_ghostwriter_message = AsyncMock(return_value={ + "entry_identifier": "task-1", + "oplog": "12", + "tags": [], + }) + self.sync._execute_query = AsyncMock(side_effect=[ + model_not_found_error(), + {"oplogEntry": [{"id": "99"}]}, + {"update_oplogEntry": {"returning": [{"id": "99"}]}}, + ]) + + await self.sync._update_entry({"agent_task_id": "task-1", "id": "mythic-row"}, "41") + + first_call = self.sync._execute_query.await_args_list[0] + self.assertFalse(first_call.kwargs["retry_model_not_found"]) + self.assertEqual(self.sync.rconn.get(self.sync._redis_entry_key("task-1")), b"99") + + async def test_failed_tag_update_remains_queued_without_blocking_entry(self): + self.sync._queue_task_tags(["mythic:test"], "99", "task-1") + self.sync._handle_task_tags = AsyncMock(side_effect=RuntimeError("temporary failure")) + + with patch("sync.random.uniform", return_value=10): + await self.sync._process_pending_tag_update("99") + + payload = self.sync.rconn.hget(self.sync.tag_retry_data_key, "99") + self.assertIsNotNone(payload) + self.assertIn(b'"attempt": 1', payload) + self.assertIn("99", self.sync.rconn.sorted_sets[self.sync.tag_retry_schedule_key]) + + async def test_successful_tag_update_is_removed_from_queue(self): + self.sync._queue_task_tags(["mythic:test"], "99", "task-1") + self.sync._handle_task_tags = AsyncMock() + + await self.sync._process_pending_tag_update("99") + + self.assertIsNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "99")) + self.assertNotIn("99", self.sync.rconn.sorted_sets[self.sync.tag_retry_schedule_key]) + + async def test_completed_tag_job_does_not_remove_newer_payload(self): + self.sync._queue_task_tags(["mythic:old"], "99", "task-1") + + async def replace_with_newer_payload(*args, **kwargs): + self.sync._queue_task_tags(["mythic:new"], "99", "task-1") + + self.sync._handle_task_tags = AsyncMock(side_effect=replace_with_newer_payload) + + await self.sync._process_pending_tag_update("99") + + payload = self.sync.rconn.hget(self.sync.tag_retry_data_key, "99") + self.assertIn(b'"mythic:new"', payload) + self.assertIn("99", self.sync.rconn.sorted_sets[self.sync.tag_retry_schedule_key]) + + async def test_missing_tag_target_moves_to_current_entry(self): + self.sync._queue_task_tags(["mythic:test"], "41", "task-1") + self.sync._handle_task_tags = AsyncMock(side_effect=model_not_found_error()) + self.sync._execute_query = AsyncMock(return_value={"oplogEntry": [{"id": "99"}]}) + + await self.sync._process_pending_tag_update("41") + + self.assertIsNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "41")) + payload = self.sync.rconn.hget(self.sync.tag_retry_data_key, "99") + self.assertIn(b'"entry_id": "99"', payload) + self.assertEqual(self.sync.rconn.get(self.sync._redis_entry_key("task-1")), b"99") + + async def test_deleted_tag_target_is_retired_instead_of_retried_forever(self): + self.sync._queue_task_tags(["mythic:test"], "41", "task-1") + self.sync._handle_task_tags = AsyncMock(side_effect=model_not_found_error()) + self.sync._execute_query = AsyncMock(return_value={"oplogEntry": []}) + + await self.sync._process_pending_tag_update("41") + + self.assertIsNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "41")) + self.assertNotIn("41", self.sync.rconn.sorted_sets[self.sync.tag_retry_schedule_key]) + dead_letter = self.sync.rconn.hget(self.sync.tag_retry_dead_letter_key, "41") + self.assertIn(b'"entry_identifier": "task-1"', dead_letter) + self.assertIn(b'"reason": "entry identifier no longer resolves in Ghostwriter"', dead_letter) + self.assertEqual( + self.sync._post_error_notification.await_args.kwargs["source"], + "mythic_sync_orphaned_tag_update", + ) + + def test_legacy_redis_mapping_is_migrated_to_scoped_key(self): + self.sync.rconn.set("task-1", "41") + + entry_id = self.sync._get_cached_entry_id("task-1") + + self.assertEqual(entry_id, b"41") + self.assertIsNone(self.sync.rconn.get("task-1")) + self.assertEqual(self.sync.rconn.get(self.sync._redis_entry_key("task-1")), b"41") + + def test_ip_only_redis_mapping_is_migrated_to_port_scoped_key(self): + legacy_key = self.sync._legacy_ip_redis_entry_key("task-1") + self.sync.rconn.set(legacy_key, "41") + + entry_id = self.sync._get_cached_entry_id("task-1") + + self.assertEqual(entry_id, b"41") + self.assertIsNone(self.sync.rconn.get(legacy_key)) + self.assertEqual(self.sync.rconn.get(self.sync._redis_entry_key("task-1")), b"41") + + def test_legacy_tag_queue_is_migrated_to_scoped_keys(self): + self.sync.rconn.hset( + self.sync.legacy_tag_retry_data_key, + "99", + '{"attempt": 1, "entry_id": "99", "tags": []}', + ) + self.sync.rconn.zadd(self.sync.legacy_tag_retry_schedule_key, {"99": 123.0}) + + migrated = self.sync._migrate_legacy_tag_queue() + + self.assertEqual(migrated, 1) + self.assertIsNotNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "99")) + self.assertEqual(self.sync.rconn.sorted_sets[self.sync.tag_retry_schedule_key]["99"], 123.0) + self.assertEqual(self.sync.rconn.hlen(self.sync.legacy_tag_retry_data_key), 0) + + def test_ip_only_tag_and_dead_letter_state_is_migrated_to_port_scope(self): + self.sync.rconn.hset( + self.sync.legacy_ip_tag_retry_data_key, + "99", + '{"attempt": 1, "entry_id": "99", "entry_identifier": "task-1", "tags": []}', + ) + self.sync.rconn.zadd(self.sync.legacy_ip_tag_retry_schedule_key, {"99": 123.0}) + self.sync.rconn.hset( + self.sync.legacy_ip_tag_retry_dead_letter_key, + "41", + '{"entry_id": "41", "entry_identifier": "task-2", "reason": "deleted"}', + ) + + migrated = self.sync._migrate_legacy_tag_queue() + + self.assertEqual(migrated, 2) + self.assertIsNotNone(self.sync.rconn.hget(self.sync.tag_retry_data_key, "99")) + self.assertEqual(self.sync.rconn.sorted_sets[self.sync.tag_retry_schedule_key]["99"], 123.0) + self.assertIsNotNone(self.sync.rconn.hget(self.sync.tag_retry_dead_letter_key, "41")) + self.assertEqual(self.sync.rconn.hlen(self.sync.legacy_ip_tag_retry_data_key), 0) + self.assertEqual(self.sync.rconn.hlen(self.sync.legacy_ip_tag_retry_dead_letter_key), 0) + + async def test_initial_entry_is_not_duplicated(self): + self.sync._execute_query = AsyncMock(return_value={"oplogEntry": [{"id": "1"}]}) + + with patch("sync.mythic.send_event_log_message", new=AsyncMock()): + await self.sync._create_initial_entry() + + self.sync._execute_query.assert_awaited_once_with( + self.sync.entry_identifier_query, + { + "oplog": self.sync.GHOSTWRITER_OPLOG_ID, + "entry_identifier": ( + f"mythic_sync:initial:{self.sync.MYTHIC_IP}:{self.sync.MYTHIC_PORT}" + ), + }, + ) + + async def test_api_key_authentication_does_not_require_user_credentials(self): + self.sync.MYTHIC_API_KEY = "mythic-api-key" + self.sync.MYTHIC_USERNAME = "" + self.sync.MYTHIC_PASSWORD = "" + mythic_instance = object() + + with patch("sync.mythic.login", new=AsyncMock(return_value=mythic_instance)) as login, patch( + "sync.mythic.get_me", new=AsyncMock() + ): + result = await self.sync._MythicSync__wait_for_authentication() + + self.assertIs(result, mythic_instance) + self.assertEqual(login.await_args.kwargs["apitoken"], "mythic-api-key") + + async def test_wait_for_redis_requires_a_successful_ping(self): + redis_connection = FakeRedis() + + with patch("sync.redis.Redis", return_value=redis_connection) as redis_client: + await self.sync._wait_for_redis() + + self.assertIs(self.sync.rconn, redis_connection) + self.assertEqual(redis_client.call_args.kwargs["db"], self.sync.REDIS_DB) + + async def test_wait_for_redis_propagates_cancellation_without_retrying(self): + with patch("sync.redis.Redis", side_effect=asyncio.CancelledError), patch( + "sync.asyncio.sleep", new_callable=AsyncMock + ) as sleep: + with self.assertRaises(asyncio.CancelledError): + await self.sync._wait_for_redis() + + sleep.assert_not_awaited() + self.sync._post_error_notification.assert_not_awaited() + + async def test_redis_startup_warns_about_embedded_storage_and_dead_letters(self): + redis_connection = FakeRedis() + redis_connection.hset( + self.sync.tag_retry_dead_letter_key, + "41", + '{"entry_identifier": "task-1", "reason": "deleted"}', + ) + self.sync.REDIS_HOSTNAME = "127.0.0.1" + + with patch("sync.redis.Redis", return_value=redis_connection), self.assertLogs( + "mythic_sync_logger", level="WARNING" + ) as logs: + await self.sync._wait_for_redis() + + output = "\n".join(logs.output) + self.assertIn("mount stable persistent storage at /data", output) + self.assertIn(self.sync.tag_retry_dead_letter_key, output) + self.assertIn("inspect with HGETALL", output) + + async def test_redis_url_supports_credentials_without_logging_them(self): + redis_connection = FakeRedis() + self.sync.REDIS_URL = "rediss://sync-user:secret-password@redis.example:6380/4" + + with patch("sync.redis.Redis.from_url", return_value=redis_connection) as from_url, self.assertLogs( + "mythic_sync_logger", level="INFO" + ) as logs: + await self.sync._wait_for_redis() + + from_url.assert_called_once_with( + self.sync.REDIS_URL, + socket_connect_timeout=5, + socket_timeout=5, + ) + output = "\n".join(logs.output) + self.assertIn("configured REDIS_URL", output) + self.assertNotIn("secret-password", output) + + async def test_error_notification_is_skipped_before_mythic_authentication(self): + self.sync.mythic_instance = None + + with patch("sync.mythic.send_event_log_message", new=AsyncMock()) as send_message: + await MythicSync._post_error_notification(self.sync, message="Redis unavailable") + + send_message.assert_not_awaited() + + async def test_create_entry_propagates_unexpected_failures(self): + self.sync._mythic_task_to_ghostwriter_message = AsyncMock(return_value={ + "entry_identifier": "task-1", + "oplog": "12", + "tags": [], + }) + self.sync._execute_query = AsyncMock(side_effect=RuntimeError("Redis or response failure")) + + with self.assertRaisesRegex(RuntimeError, "Redis or response failure"): + await self.sync._create_entry({"agent_task_id": "task-1"}) + + self.sync._post_error_notification.assert_awaited_once() + + async def test_create_entry_propagates_cancellation_without_notification(self): + self.sync._mythic_task_to_ghostwriter_message = AsyncMock( + side_effect=asyncio.CancelledError + ) + + with self.assertRaises(asyncio.CancelledError): + await self.sync._create_entry({"agent_task_id": "task-1"}) + + self.sync._post_error_notification.assert_not_awaited() + + +if __name__ == "__main__": + unittest.main()