mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-25 23:53:42 +01:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2a858a2a05 | ||
|
|
4a7100e0c9 |
No files matched your search
@@ -14,10 +14,10 @@ env:
|
||||
REGISTRY: ghcr.io
|
||||
IMAGE_NAME: ${{ github.repository }}
|
||||
|
||||
# There is a single job in this workflow. It's configured to run on the latest available version of Ubuntu.
|
||||
# Build each architecture natively, then publish the combined image manifests.
|
||||
jobs:
|
||||
build-and-push-image:
|
||||
runs-on: ubuntu-latest
|
||||
runs-on: ${{ matrix.architecture.runner }}
|
||||
# Sets the permissions granted to the `GITHUB_TOKEN` for the actions in this job.
|
||||
permissions:
|
||||
contents: read
|
||||
@@ -26,7 +26,15 @@ jobs:
|
||||
id-token: write
|
||||
#
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
architecture:
|
||||
- platform: linux/amd64
|
||||
runner: ubuntu-24.04
|
||||
name: amd64
|
||||
- platform: linux/arm64
|
||||
runner: ubuntu-24.04-arm
|
||||
name: arm64
|
||||
platform:
|
||||
- dockerfile: "Dockerfile"
|
||||
tag-suffix: ""
|
||||
@@ -61,24 +69,95 @@ jobs:
|
||||
type=raw,enable=${{ github.ref_type == 'tag' }},value=${{ github.ref_name }},suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=sha,suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
|
||||
# Register emulators so Buildx can build the ARM64 images on the AMD64 runner.
|
||||
- name: Set up QEMU
|
||||
uses: docker/setup-qemu-action@v4
|
||||
# This step sets up some additional capabilities to generate the provenance and sbom attestations
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v4
|
||||
# This step uses the `docker/build-push-action` action to build the image, based on your repository's `Dockerfile`. If the build succeeds, it pushes the image to GitHub Packages.
|
||||
# It uses the `context` parameter to define the build's context as the set of files located in the specified path. For more information, see "[Usage](https://github.com/docker/build-push-action#usage)" in the README of the `docker/build-push-action` repository.
|
||||
# It uses the `tags` and `labels` parameters to tag and label the image with the output from the "meta" step.
|
||||
# Push by digest so architecture builds do not overwrite shared image tags.
|
||||
- name: Build and push Docker image
|
||||
id: push
|
||||
uses: docker/build-push-action@v7
|
||||
with:
|
||||
file: ${{ matrix.platform.dockerfile }}
|
||||
context: .
|
||||
platforms: linux/amd64,linux/arm64
|
||||
push: true
|
||||
tags: ${{ steps.meta.outputs.tags }}
|
||||
platforms: ${{ matrix.architecture.platform }}
|
||||
outputs: type=image,name=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
|
||||
labels: ${{ steps.meta.outputs.labels }}
|
||||
provenance: mode=max
|
||||
sbom: true
|
||||
- name: Export image digest
|
||||
env:
|
||||
DIGEST: ${{ steps.push.outputs.digest }}
|
||||
run: |
|
||||
mkdir -p "${RUNNER_TEMP}/digests"
|
||||
touch "${RUNNER_TEMP}/digests/${DIGEST#sha256:}"
|
||||
- name: Upload image digest
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: digests-${{ matrix.platform.dockerfile }}-${{ matrix.architecture.name }}
|
||||
path: ${{ runner.temp }}/digests/*
|
||||
if-no-files-found: error
|
||||
retention-days: 1
|
||||
|
||||
publish-manifests:
|
||||
needs: build-and-push-image
|
||||
runs-on: ubuntu-24.04
|
||||
permissions:
|
||||
contents: read
|
||||
packages: write
|
||||
strategy:
|
||||
matrix:
|
||||
platform:
|
||||
- dockerfile: "Dockerfile"
|
||||
tag-suffix: ""
|
||||
- dockerfile: "Dockerfile.ios"
|
||||
tag-suffix: "-ios"
|
||||
- dockerfile: "Dockerfile.android"
|
||||
tag-suffix: "-android"
|
||||
steps:
|
||||
- name: Download image digests
|
||||
uses: actions/download-artifact@v4
|
||||
with:
|
||||
pattern: digests-${{ matrix.platform.dockerfile }}-*
|
||||
path: ${{ runner.temp }}/digests
|
||||
merge-multiple: true
|
||||
- name: Log in to the Container registry
|
||||
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
|
||||
with:
|
||||
registry: ${{ env.REGISTRY }}
|
||||
username: ${{ github.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
- name: Extract metadata (tags, labels) for Docker
|
||||
id: meta
|
||||
uses: docker/metadata-action@v6
|
||||
with:
|
||||
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
|
||||
flavor: |
|
||||
latest=false
|
||||
tags: |
|
||||
type=raw,value=latest,enable={{ is_default_branch }},suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=raw,enable=${{ github.event_name == 'release' || github.ref_type == 'tag' }},value=stable,suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=raw,enable=${{ github.event_name == 'release' }},value=${{ github.event.release.tag_name }},suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=raw,enable=${{ github.ref_type == 'tag' }},value=${{ github.ref_name }},suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=sha,suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v4
|
||||
- name: Publish multi-platform image
|
||||
working-directory: ${{ runner.temp }}/digests
|
||||
env:
|
||||
METADATA_JSON: ${{ steps.meta.outputs.json }}
|
||||
run: |
|
||||
mapfile -t tags < <(jq -r '.tags[]' <<< "$METADATA_JSON")
|
||||
tag_args=()
|
||||
for tag in "${tags[@]}"; do
|
||||
tag_args+=(--tag "$tag")
|
||||
done
|
||||
images=()
|
||||
for digest in *; do
|
||||
images+=("${REGISTRY}/${IMAGE_NAME}@sha256:${digest}")
|
||||
done
|
||||
test "${#images[@]}" -eq 2
|
||||
docker buildx imagetools create "${tag_args[@]}" "${images[@]}"
|
||||
docker buildx imagetools inspect "${tags[0]}"
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
# Installation
|
||||
|
||||
Before proceeding, please note that MVT requires Python 3.10 or newer to run. While it should be available on most operating systems, please make sure of that before proceeding.
|
||||
Before proceeding, please note that MVT requires Python 3.6+ to run. While it should be available on most operating systems, please make sure of that before proceeding.
|
||||
|
||||
## Dependencies on Linux
|
||||
|
||||
|
||||
@@ -204,10 +204,6 @@ This JSON file is created by mvt-ios' `Manifest` module. The module extracts rec
|
||||
|
||||
If indicators are provided through the command-line, they are checked against the original relative path in case. In some cases, there might be records of files created containing a domain name in their name, for example in the case of browser cache folders. Any matches are stored in *manifest_detected.json*.
|
||||
|
||||
An incomplete backup still lists in its manifest the files it failed to acquire. Those records carry a `"missing": true` field, and the module reports how many of them it found. Use it to tell a module which returned nothing because the artifact was not acquired apart from one which found nothing on a device that never had it.
|
||||
|
||||
Files must be stored at their expected paths inside the backup folder. If the module cannot finish listing the backup files, it logs a warning and skips the missing-file check; the manifest metadata is still extracted.
|
||||
|
||||
---
|
||||
|
||||
### `os_analytics_ad_daily.json`
|
||||
|
||||
@@ -29,21 +29,11 @@ class DumpsysADBArtifact(AndroidArtifact):
|
||||
stack = [res]
|
||||
cur_indent = 0
|
||||
in_multiline = False
|
||||
for line in dump_data.strip(b"\r\n").split(b"\n"):
|
||||
line = line.removesuffix(b"\r")
|
||||
for line in dump_data.strip(b"\n").split(b"\n"):
|
||||
# Track the level of indentation
|
||||
indent = len(line) - len(line.lstrip())
|
||||
if indent < cur_indent:
|
||||
# If the current line is less indented than the previous one, back out
|
||||
if len(stack) <= 1:
|
||||
# Dedenting below the outermost level means this is not the
|
||||
# well-formed block the parser expects. Stop here rather than
|
||||
# raise IndexError out of the module on the next line.
|
||||
self.log.error(
|
||||
"Unexpected indentation in ADB manager state, "
|
||||
"stopping the parse of this section"
|
||||
)
|
||||
break
|
||||
stack.pop()
|
||||
cur_indent = indent
|
||||
else:
|
||||
@@ -73,12 +63,6 @@ class DumpsysADBArtifact(AndroidArtifact):
|
||||
current_dict = stack[-1]
|
||||
|
||||
if key == "}":
|
||||
if len(stack) <= 1:
|
||||
self.log.error(
|
||||
"Unbalanced closing brace in ADB manager state, "
|
||||
"stopping the parse of this section"
|
||||
)
|
||||
break
|
||||
stack.pop()
|
||||
continue
|
||||
|
||||
@@ -177,26 +161,6 @@ class DumpsysADBArtifact(AndroidArtifact):
|
||||
f"'{user_key['fingerprint']}'"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _find_state_end(content: bytes, open_brace: int) -> int:
|
||||
"""Find the unindented line closing the ADB manager state, or -1.
|
||||
|
||||
Braces inside key comments or embedded keystore data are values, while
|
||||
nested structural closing braces are indented.
|
||||
"""
|
||||
line_start = open_brace
|
||||
while line_start < len(content):
|
||||
line_end = content.find(b"\n", line_start)
|
||||
if line_end == -1:
|
||||
line_end = len(content)
|
||||
line = content[line_start:line_end].removesuffix(b"\r")
|
||||
if line_start > open_brace and line == b"}":
|
||||
return line_start
|
||||
if line.startswith((b"---------", b"DUMP OF SERVICE ")):
|
||||
break
|
||||
line_start = line_end + 1
|
||||
return -1
|
||||
|
||||
def parse(self, content: bytes) -> None:
|
||||
"""
|
||||
Parse the Dumpsys ADB section
|
||||
@@ -216,16 +180,12 @@ class DumpsysADBArtifact(AndroidArtifact):
|
||||
self.log.error("Unable to find ADB manager state in dumpsys output")
|
||||
return
|
||||
|
||||
end_of_json = self._find_state_end(content, start_of_json + 1)
|
||||
if end_of_json == -1:
|
||||
self.log.error(
|
||||
"Unable to find complete ADB manager state in dumpsys output"
|
||||
)
|
||||
end_of_json = content.rfind(b"}\n")
|
||||
if end_of_json == -1 or end_of_json <= start_of_json:
|
||||
self.log.error("Unable to find complete ADB manager state in dumpsys output")
|
||||
return
|
||||
|
||||
# The brace that opens the state and the one that closes it are not part
|
||||
# of the indented body.
|
||||
json_content = content[start_of_json + 2 : end_of_json].rstrip()
|
||||
json_content = content[start_of_json + 2 : end_of_json - 2].rstrip()
|
||||
|
||||
parsed = self.indented_dump_parser(json_content)
|
||||
if parsed.get("debugging_manager") is None:
|
||||
|
||||
@@ -73,12 +73,8 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
|
||||
continue
|
||||
|
||||
line = line.strip().replace("Update ", "")
|
||||
# A truncated or vendor-specific line must not abort the parse and
|
||||
# lose every record after it.
|
||||
package_name, _, vers = line.partition(" ")
|
||||
vers_raw = vers.partition("=")[2]
|
||||
if not package_name or not vers_raw:
|
||||
continue
|
||||
package_name, vers = line.split(" ", 1)
|
||||
vers_raw = vers.split("=", 1)[1]
|
||||
try:
|
||||
version_code: int | str = int(vers_raw)
|
||||
except ValueError:
|
||||
|
||||
@@ -1279,9 +1279,5 @@
|
||||
{
|
||||
"version": "27.0",
|
||||
"build": "24A437"
|
||||
},
|
||||
{
|
||||
"version": "27.0.1",
|
||||
"build": "24A446"
|
||||
}
|
||||
]
|
||||
@@ -147,12 +147,6 @@ class Manifest(IOSExtraction):
|
||||
)
|
||||
names = [description[0] for description in cur.description]
|
||||
|
||||
# An incomplete backup still lists the files it failed to acquire in
|
||||
# its manifest. Only regular files are stored in the backup folder, so
|
||||
# directories and symlinks (flags 2 and 4) are not looked up.
|
||||
stored_file_ids = self._get_stored_backup_file_ids()
|
||||
missing_files = 0
|
||||
|
||||
for file_entry in cur:
|
||||
file_data = {}
|
||||
for index, value in enumerate(file_entry):
|
||||
@@ -166,22 +160,6 @@ class Manifest(IOSExtraction):
|
||||
"created": "",
|
||||
}
|
||||
|
||||
if (
|
||||
stored_file_ids is not None
|
||||
and file_data["flags"] == 1
|
||||
and file_data["fileID"] not in stored_file_ids
|
||||
):
|
||||
# Without this, a module which found nothing for one of these
|
||||
# files would look like a negative result rather than a gap in
|
||||
# the acquisition.
|
||||
cleaned_metadata["missing"] = True
|
||||
missing_files += 1
|
||||
self.log.debug(
|
||||
"File %s is listed in the manifest but was not found in the "
|
||||
"backup folder",
|
||||
cleaned_metadata["relative_path"],
|
||||
)
|
||||
|
||||
if file_data["file"]:
|
||||
try:
|
||||
file_plist = plistlib.load(io.BytesIO(file_data["file"]))
|
||||
@@ -219,10 +197,3 @@ class Manifest(IOSExtraction):
|
||||
conn.close()
|
||||
|
||||
self.log.info("Extracted a total of %d file metadata items", len(self.results))
|
||||
|
||||
if missing_files:
|
||||
self.log.info(
|
||||
"Found %d files listed in the manifest but missing from the backup "
|
||||
"folder. The backup might be incomplete.",
|
||||
missing_files,
|
||||
)
|
||||
@@ -216,55 +216,6 @@ class IOSExtraction(MVTModule):
|
||||
|
||||
return None
|
||||
|
||||
def _get_stored_backup_file_ids(self) -> Optional[set[str]]:
|
||||
"""List the IDs of the files actually stored in the backup folder.
|
||||
|
||||
A backup folder stores each file under a two character subfolder named
|
||||
after the first two characters of its file ID. Walking the folders once
|
||||
is cheaper than a filesystem lookup per file ID, and it keeps callers
|
||||
which compare a whole manifest against the folder from paying a
|
||||
`resolve()` for every entry.
|
||||
|
||||
:returns: The file IDs found at their expected paths within the backup
|
||||
folder, or None if the inventory could not be completed.
|
||||
"""
|
||||
if not self.target_path:
|
||||
return None
|
||||
|
||||
file_ids: set[str] = set()
|
||||
try:
|
||||
backup_root = Path(self.target_path).resolve()
|
||||
with os.scandir(self.target_path) as entries:
|
||||
for entry in entries:
|
||||
if len(entry.name) != 2 or any(
|
||||
char not in "0123456789abcdef" for char in entry.name
|
||||
):
|
||||
continue
|
||||
if not entry.is_dir():
|
||||
continue
|
||||
if not Path(entry.path).resolve().is_relative_to(backup_root):
|
||||
continue
|
||||
with os.scandir(entry.path) as sub_entries:
|
||||
for sub_entry in sub_entries:
|
||||
if sub_entry.name[:2] != entry.name:
|
||||
continue
|
||||
if sub_entry.is_symlink() and not Path(
|
||||
sub_entry.path
|
||||
).resolve().is_relative_to(backup_root):
|
||||
continue
|
||||
if sub_entry.is_file():
|
||||
file_ids.add(sub_entry.name)
|
||||
except OSError as exc:
|
||||
self.log.warning(
|
||||
"Unable to list the files stored in the backup folder %s: %s. "
|
||||
"Skipping the missing-file check.",
|
||||
self.target_path,
|
||||
exc,
|
||||
)
|
||||
return None
|
||||
|
||||
return file_ids
|
||||
|
||||
def _get_fs_files_from_patterns(self, root_paths: list) -> Iterator[str]:
|
||||
if not self.target_path:
|
||||
return
|
||||
|
||||
@@ -82,9 +82,7 @@ class Filesystem(IOSExtraction):
|
||||
try:
|
||||
dir_path = os.path.join(root, dir_name)
|
||||
result = {
|
||||
"path": os.path.relpath(dir_path, self.target_path).replace(
|
||||
os.sep, "/"
|
||||
),
|
||||
"path": os.path.relpath(dir_path, self.target_path),
|
||||
"modified": convert_unix_to_iso(os.stat(dir_path).st_mtime),
|
||||
}
|
||||
except Exception:
|
||||
@@ -96,9 +94,7 @@ class Filesystem(IOSExtraction):
|
||||
try:
|
||||
file_path = os.path.join(root, file_name)
|
||||
result = {
|
||||
"path": os.path.relpath(file_path, self.target_path).replace(
|
||||
os.sep, "/"
|
||||
),
|
||||
"path": os.path.relpath(file_path, self.target_path),
|
||||
"modified": convert_unix_to_iso(os.stat(file_path).st_mtime),
|
||||
}
|
||||
except Exception:
|
||||
|
||||
@@ -130,101 +130,6 @@ class TestDumpsysADBArtifact:
|
||||
assert key_store_entry["fingerprint"] == expected_fingerprint
|
||||
assert key_store_entry["last_connected"] == "1628501829898"
|
||||
|
||||
def test_parsing_adb_xml_with_crlf_line_endings(self):
|
||||
da_adb = DumpsysADBArtifact()
|
||||
file = get_artifact("android_data/dumpsys_adb_xml.txt")
|
||||
with open(file, "rb") as f:
|
||||
data = f.read().replace(b"\r\n", b"\n").replace(b"\n", b"\r\n")
|
||||
|
||||
da_adb.parse(data)
|
||||
|
||||
assert len(da_adb.results) == 1
|
||||
assert da_adb.results[0]["user_keys"][0]["user"] == "user@laptop"
|
||||
assert da_adb.results[0]["keystore"][0]["last_connected"] == "1628501829898"
|
||||
|
||||
def test_parsing_adb_wifi_with_mixed_line_endings(self):
|
||||
da_adb = DumpsysADBArtifact()
|
||||
data = (
|
||||
b"ADB MANAGER STATE (dumpsys adb):\n"
|
||||
b"{\n"
|
||||
b" debugging_manager={\n"
|
||||
b" connected_to_adb=true\n"
|
||||
b" user_keys=QUJDRA== host@example\n"
|
||||
b" adb_wifi={\n"
|
||||
b" enabled=false\n"
|
||||
b" }\n"
|
||||
b" }\r\n"
|
||||
b"}\n"
|
||||
b"--------- duration\n"
|
||||
)
|
||||
|
||||
da_adb.parse(data)
|
||||
|
||||
assert len(da_adb.results) == 1
|
||||
assert da_adb.results[0]["user_keys"][0]["user"] == "host@example"
|
||||
assert da_adb.results[0]["adb_wifi"]["enabled"] == b"false"
|
||||
|
||||
ADB_STATE = (
|
||||
b"ADB MANAGER STATE (dumpsys adb):\n"
|
||||
b"{\n"
|
||||
b" debugging_manager={\n"
|
||||
b" connected_to_adb=true\n"
|
||||
b" user_keys=QUJDRA== host@example\n"
|
||||
b" }\n"
|
||||
b"}\n"
|
||||
b"--------- 0.5s was the duration of 'dumpsys adb'\n"
|
||||
)
|
||||
|
||||
def test_a_later_dumpsys_section_does_not_extend_the_adb_state(self):
|
||||
# A bug report holds many sections. Looking for the last closing brace
|
||||
# in the whole output pulled a later section into this one, which threw
|
||||
# IndexError out of parse() and lost the ADB records entirely.
|
||||
da_adb = DumpsysADBArtifact()
|
||||
da_adb.parse(
|
||||
self.ADB_STATE + b"DUMP OF SERVICE other:\n"
|
||||
b" debugging_manager={\n"
|
||||
b" connected_to_adb=false\n"
|
||||
b" user_keys=RVZJTA== attacker@host\n"
|
||||
b" }\n"
|
||||
b"}\n"
|
||||
)
|
||||
|
||||
assert len(da_adb.results) == 1
|
||||
assert [key["user"] for key in da_adb.results[0]["user_keys"]] == [
|
||||
"host@example"
|
||||
]
|
||||
assert da_adb.results[0]["connected_to_adb"] is True
|
||||
|
||||
def test_braces_in_key_comment_are_not_state_delimiters(self):
|
||||
for user in (b"host{example", b"host}}example"):
|
||||
da_adb = DumpsysADBArtifact()
|
||||
da_adb.parse(
|
||||
b"ADB MANAGER STATE (dumpsys adb):\n"
|
||||
b"{\n"
|
||||
b" debugging_manager={\n"
|
||||
b" connected_to_adb=true\n"
|
||||
b" user_keys=QUJDRA== " + user + b"\n }\n}\n"
|
||||
)
|
||||
|
||||
assert len(da_adb.results) == 1
|
||||
assert da_adb.results[0]["user_keys"][0]["user"] == user.decode()
|
||||
|
||||
def test_unbalanced_state_is_reported_rather_than_raising(self):
|
||||
da_adb = DumpsysADBArtifact()
|
||||
da_adb.parse(
|
||||
b"ADB MANAGER STATE (dumpsys adb):\n"
|
||||
b"{\n"
|
||||
b" debugging_manager={\n"
|
||||
b" connected_to_adb=true\n"
|
||||
b" }\n"
|
||||
b" }\n"
|
||||
b"}\n"
|
||||
)
|
||||
|
||||
# No exception, and whatever was read before the bad line is kept.
|
||||
assert len(da_adb.results) == 1
|
||||
assert da_adb.results[0]["connected_to_adb"] is True
|
||||
|
||||
|
||||
class TestDumpsysADBStateAlerts:
|
||||
def test_no_androidqf_context_preserves_existing_behavior(self):
|
||||
|
||||
@@ -144,18 +144,3 @@ class TestDumpsysBatteryDailyArtifact:
|
||||
"Detected uninstall of package com.example.app (vers 0)"
|
||||
)
|
||||
assert uninstall_alert.event_time == "2026-01-10"
|
||||
|
||||
def test_malformed_update_line_does_not_lose_later_records(self):
|
||||
# A truncated "Update" line, or one from a vendor that omits "vers=",
|
||||
# used to raise out of parse() and lose every record after it.
|
||||
dba = DumpsysBatteryDailyArtifact()
|
||||
dba.parse(
|
||||
" Daily from 2021-05-10-08-00-00 to 2021-05-11-08-00-00:\n"
|
||||
" Update com.first vers=1\n"
|
||||
" Update com.truncated\n"
|
||||
" Update com.no.equals vers 2\n"
|
||||
" Update com.last vers=3\n"
|
||||
)
|
||||
|
||||
assert [r["package_name"] for r in dba.results] == ["com.first", "com.last"]
|
||||
assert [r["version_code"] for r in dba.results] == [1, 3]
|
||||
@@ -1,4 +1,3 @@
|
||||
import shlex
|
||||
from types import SimpleNamespace
|
||||
|
||||
import click
|
||||
@@ -111,7 +110,7 @@ def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
|
||||
def test_loaded_command_participates_in_shell_completion(tmp_path):
|
||||
command_path = _write_command(tmp_path / "hello.py", "hello")
|
||||
group = _make_group()
|
||||
words = f"group --load-command {shlex.quote(str(command_path))} he"
|
||||
words = f"group --load-command {command_path} he"
|
||||
|
||||
result = CliRunner().invoke(
|
||||
group,
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import json
|
||||
import re
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
@@ -55,23 +54,20 @@ def _run(command, arguments):
|
||||
return CliRunner().invoke(command, arguments, env={"COLUMNS": "200"})
|
||||
|
||||
|
||||
def _table_lines(output):
|
||||
"""Return the cells of each line of the table, header first."""
|
||||
def _table_rows(output):
|
||||
"""Return the content of the table rows, without the header and the box."""
|
||||
return [
|
||||
[cell.strip() for cell in re.split("[│┃]", line.strip().strip("│┃"))]
|
||||
[cell.strip() for cell in line.strip().strip("│").split("│")]
|
||||
for line in output.splitlines()
|
||||
if "│" in line or "┃" in line
|
||||
if "│" in line
|
||||
]
|
||||
|
||||
|
||||
def _table_rows(output):
|
||||
"""Return the content of the table rows, without the header and the box."""
|
||||
return _table_lines(output)[1:]
|
||||
|
||||
|
||||
def _table_header(output):
|
||||
lines = _table_lines(output)
|
||||
return lines[0] if lines else []
|
||||
for line in output.splitlines():
|
||||
if "┃" in line:
|
||||
return [cell.strip() for cell in line.strip().strip("┃").split("┃")]
|
||||
return []
|
||||
|
||||
|
||||
def _install(monkeypatch, distributions, entry_points):
|
||||
|
||||
@@ -7,7 +7,6 @@ import logging
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from Crypto.Cipher import AES
|
||||
|
||||
from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup
|
||||
@@ -82,10 +81,7 @@ def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
|
||||
assert output_path.read_bytes() == b"plain content"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("with_symlink", [False, True], ids=["file-id", "symlink"])
|
||||
def test_process_backup_rejects_unsafe_file_ids_and_destinations(
|
||||
mocker, tmp_path, with_symlink
|
||||
):
|
||||
def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path):
|
||||
backup_path = tmp_path / "backup"
|
||||
destination = tmp_path / "destination"
|
||||
outside = tmp_path / "outside"
|
||||
@@ -96,27 +92,20 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(
|
||||
safe_file_id = "ef" + "3" * 38
|
||||
unsafe_file_id = "../../outside-file"
|
||||
symlink_file_id = "ab" + "4" * 38
|
||||
file_ids = [safe_file_id]
|
||||
if with_symlink:
|
||||
file_ids.append(symlink_file_id)
|
||||
for file_id in file_ids:
|
||||
for file_id in (safe_file_id, symlink_file_id):
|
||||
source_path = backup_path / file_id[:2] / file_id
|
||||
source_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
source_path.write_bytes(b"encrypted")
|
||||
if with_symlink:
|
||||
try:
|
||||
(destination / "ab").symlink_to(outside, target_is_directory=True)
|
||||
except OSError:
|
||||
pytest.skip("creating symbolic links is not permitted on this system")
|
||||
(destination / "ab").symlink_to(outside, target_is_directory=True)
|
||||
|
||||
cursor = mocker.MagicMock()
|
||||
records = [
|
||||
(safe_file_id, "Domain", "safe", b"plist"),
|
||||
(unsafe_file_id, "Domain", "unsafe", b"plist"),
|
||||
]
|
||||
if with_symlink:
|
||||
records.append((symlink_file_id, "Domain", "symlink", b"plist"))
|
||||
cursor.__iter__.return_value = iter(records)
|
||||
cursor.__iter__.return_value = iter(
|
||||
[
|
||||
(safe_file_id, "Domain", "safe", b"plist"),
|
||||
(unsafe_file_id, "Domain", "unsafe", b"plist"),
|
||||
(symlink_file_id, "Domain", "symlink", b"plist"),
|
||||
]
|
||||
)
|
||||
cursor_context = mocker.MagicMock()
|
||||
cursor_context.__enter__.return_value = cursor
|
||||
|
||||
@@ -135,8 +124,7 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(
|
||||
decryptor.process_backup()
|
||||
|
||||
assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted"
|
||||
if with_symlink:
|
||||
assert not (outside / symlink_file_id).exists()
|
||||
assert not (outside / symlink_file_id).exists()
|
||||
backup.extract_file_by_id.assert_called_once()
|
||||
assert backup.extract_file_by_id.call_args.kwargs["file_id"] == safe_file_id
|
||||
|
||||
|
||||
@@ -5,12 +5,7 @@
|
||||
|
||||
import gc
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import warnings
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import run_module
|
||||
@@ -19,58 +14,8 @@ from mvt.ios.modules.backup.manifest import Manifest
|
||||
|
||||
from ..utils import get_ios_backup_folder
|
||||
|
||||
# fileID of HomeDomain::Library/SMS/sms.db in the test backup. It is one of the
|
||||
# few files the test backup actually stores.
|
||||
SMS_FILE_ID = "3d0d7e5fb2ce288813306e4d4636395e047a3d28"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def backup_without_stored_files(tmp_path):
|
||||
"""A copy of the test backup with every stored file removed.
|
||||
|
||||
The test backup only ships a handful of the files its manifest lists, so
|
||||
dropping what it does store leaves a backup where every regular file the
|
||||
manifest mentions is missing from the folder.
|
||||
"""
|
||||
backup_path = tmp_path / "backup"
|
||||
shutil.copytree(get_ios_backup_folder(), backup_path)
|
||||
|
||||
for file_id_folder in backup_path.iterdir():
|
||||
if not file_id_folder.is_dir():
|
||||
continue
|
||||
for backup_file in file_id_folder.iterdir():
|
||||
backup_file.unlink()
|
||||
|
||||
return str(backup_path)
|
||||
|
||||
|
||||
class TestIOSExtraction:
|
||||
@pytest.mark.parametrize("link_directory", [False, True], ids=["file", "directory"])
|
||||
@pytest.mark.parametrize("inside_backup", [False, True], ids=["outside", "inside"])
|
||||
def test_stored_file_inventory_matches_symlink_resolution(
|
||||
self, tmp_path, link_directory, inside_backup
|
||||
):
|
||||
backup_path = tmp_path / "backup"
|
||||
backup_path.mkdir()
|
||||
destination = (backup_path if inside_backup else tmp_path) / "contents"
|
||||
destination.mkdir()
|
||||
(destination / SMS_FILE_ID).write_bytes(b"backup file")
|
||||
bucket = backup_path / SMS_FILE_ID[:2]
|
||||
try:
|
||||
if link_directory:
|
||||
bucket.symlink_to(destination, target_is_directory=True)
|
||||
else:
|
||||
bucket.mkdir()
|
||||
(bucket / SMS_FILE_ID).symlink_to(destination / SMS_FILE_ID)
|
||||
except OSError:
|
||||
pytest.skip("creating symbolic links is not permitted on this system")
|
||||
|
||||
m = IOSExtraction(target_path=str(backup_path))
|
||||
assert bool(m._get_backup_file_from_id(SMS_FILE_ID)) is inside_backup
|
||||
assert m._get_stored_backup_file_ids() == (
|
||||
{SMS_FILE_ID} if inside_backup else set()
|
||||
)
|
||||
|
||||
def test_get_backup_files_from_manifest_closes_connection(self):
|
||||
m = IOSExtraction(target_path=get_ios_backup_folder())
|
||||
|
||||
@@ -96,103 +41,6 @@ class TestManifestModule:
|
||||
assert len(m.timeline) == 5881
|
||||
assert len(m.alertstore.alerts) == 0
|
||||
|
||||
def test_manifest_flags_the_files_missing_from_an_incomplete_backup(self):
|
||||
m = Manifest(target_path=get_ios_backup_folder())
|
||||
run_module(m)
|
||||
|
||||
missing = [result for result in m.results if result.get("missing")]
|
||||
# The test backup only stores a handful of the files its manifest
|
||||
# lists, and every one of the rest is a regular file (flags 1).
|
||||
assert len(missing) == 1079
|
||||
assert all(result["flags"] == 1 for result in missing)
|
||||
|
||||
stored = [result for result in m.results if result["file_id"] == SMS_FILE_ID]
|
||||
assert len(stored) == 1
|
||||
assert "missing" not in stored[0]
|
||||
|
||||
def test_manifest_flags_every_stored_file_removed_from_the_backup_folder(
|
||||
self, backup_without_stored_files
|
||||
):
|
||||
m = Manifest(target_path=backup_without_stored_files)
|
||||
run_module(m)
|
||||
|
||||
missing = [result for result in m.results if result.get("missing")]
|
||||
assert len(missing) == 1089
|
||||
|
||||
def test_manifest_flags_a_file_removed_from_the_backup_folder(self, tmp_path):
|
||||
backup_path = tmp_path / "backup"
|
||||
shutil.copytree(get_ios_backup_folder(), backup_path)
|
||||
(backup_path / SMS_FILE_ID[:2] / SMS_FILE_ID).unlink()
|
||||
|
||||
m = Manifest(target_path=str(backup_path))
|
||||
run_module(m)
|
||||
|
||||
removed = [result for result in m.results if result["file_id"] == SMS_FILE_ID]
|
||||
assert len(removed) == 1
|
||||
assert removed[0]["missing"] is True
|
||||
|
||||
@pytest.mark.parametrize("failed_folder", ["", "3d"], ids=["root", "bucket"])
|
||||
def test_manifest_skips_missing_check_when_inventory_fails(
|
||||
self, monkeypatch, caplog, failed_folder
|
||||
):
|
||||
backup_path = Path(get_ios_backup_folder())
|
||||
failed_path = backup_path / failed_folder
|
||||
scandir = os.scandir
|
||||
|
||||
def fail_listing(path):
|
||||
if Path(path) == failed_path:
|
||||
raise PermissionError("cannot list backup folder")
|
||||
return scandir(path)
|
||||
|
||||
monkeypatch.setattr(os, "scandir", fail_listing)
|
||||
m = Manifest(target_path=str(backup_path))
|
||||
with caplog.at_level(logging.INFO):
|
||||
m.run()
|
||||
|
||||
assert len(m.results) == 3721
|
||||
assert m._get_backup_file_from_id(SMS_FILE_ID) is not None
|
||||
assert all("missing" not in result for result in m.results)
|
||||
assert "Skipping the missing-file check" in caplog.text
|
||||
assert "The backup might be incomplete" not in caplog.text
|
||||
|
||||
def test_manifest_ignores_unreadable_unrelated_folders(self, tmp_path, monkeypatch):
|
||||
backup_path = tmp_path / "backup"
|
||||
shutil.copytree(get_ios_backup_folder(), backup_path)
|
||||
unrelated = backup_path / "notes"
|
||||
unrelated.mkdir()
|
||||
scandir = os.scandir
|
||||
|
||||
def fail_listing(path):
|
||||
if Path(path) == unrelated:
|
||||
raise PermissionError("cannot list unrelated folder")
|
||||
return scandir(path)
|
||||
|
||||
monkeypatch.setattr(os, "scandir", fail_listing)
|
||||
m = Manifest(target_path=str(backup_path))
|
||||
m.run()
|
||||
|
||||
assert sum(bool(result.get("missing")) for result in m.results) == 1079
|
||||
stored = next(
|
||||
result for result in m.results if result["file_id"] == SMS_FILE_ID
|
||||
)
|
||||
assert "missing" not in stored
|
||||
|
||||
@pytest.mark.parametrize("wrong_folder", ["ab", "notes"])
|
||||
def test_manifest_flags_files_in_the_wrong_folder(self, tmp_path, wrong_folder):
|
||||
backup_path = tmp_path / "backup"
|
||||
shutil.copytree(get_ios_backup_folder(), backup_path)
|
||||
destination = backup_path / wrong_folder
|
||||
destination.mkdir(exist_ok=True)
|
||||
(backup_path / SMS_FILE_ID[:2] / SMS_FILE_ID).rename(destination / SMS_FILE_ID)
|
||||
|
||||
m = Manifest(target_path=str(backup_path))
|
||||
m.run()
|
||||
|
||||
assert m._get_backup_file_from_id(SMS_FILE_ID) is None
|
||||
moved = next(result for result in m.results if result["file_id"] == SMS_FILE_ID)
|
||||
assert moved["missing"] is True
|
||||
assert sum(bool(result.get("missing")) for result in m.results) == 1080
|
||||
|
||||
def test_detection(self, indicator_file):
|
||||
m = Manifest(target_path=get_ios_backup_folder())
|
||||
ind = Indicators(log=logging.getLogger())
|
||||
|
||||
@@ -9,8 +9,6 @@ import plistlib
|
||||
import shutil
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.ios.modules.base import IOSExtraction
|
||||
from mvt.ios.modules.fs.analytics import Analytics
|
||||
|
||||
@@ -47,9 +45,6 @@ def test_open_sqlite_reads_wal_without_modifying_evidence(tmp_path):
|
||||
assert not os.path.exists(str(evidence_path) + "-shm")
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
shutil.which("sqlite3") is None, reason="the recovery needs the sqlite3 binary"
|
||||
)
|
||||
def test_recovery_preserves_source_database(tmp_path):
|
||||
database_path = tmp_path / "source.db"
|
||||
conn = sqlite3.connect(database_path)
|
||||
|
||||
@@ -3,45 +3,15 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
import logging
|
||||
import ntpath
|
||||
from types import SimpleNamespace
|
||||
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import run_module
|
||||
from mvt.ios.modules.fs import filesystem
|
||||
from mvt.ios.modules.fs.filesystem import Filesystem
|
||||
|
||||
from ..utils import get_ios_backup_folder
|
||||
|
||||
|
||||
class TestFilesystem:
|
||||
def test_windows_paths_are_normalized_for_indicators(
|
||||
self, monkeypatch, indicators_factory
|
||||
):
|
||||
m = Filesystem(target_path=r"C:\dump")
|
||||
m.indicators = indicators_factory(
|
||||
file_paths=["matched/directory"], processes=["matched"]
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
filesystem,
|
||||
"os",
|
||||
SimpleNamespace(
|
||||
sep="\\",
|
||||
path=ntpath,
|
||||
walk=lambda _: [(r"C:\dump\matched", ["directory"], ["file.txt"])],
|
||||
stat=lambda _: SimpleNamespace(st_mtime=0),
|
||||
),
|
||||
)
|
||||
|
||||
m.run()
|
||||
m.check_indicators()
|
||||
|
||||
assert {result["path"] for result in m.results} == {
|
||||
"matched/directory",
|
||||
"matched/file.txt",
|
||||
}
|
||||
assert len(m.alertstore.alerts) == 3
|
||||
|
||||
def test_filesystem(self):
|
||||
m = Filesystem(target_path=get_ios_backup_folder())
|
||||
run_module(m)
|
||||
|
||||
@@ -85,7 +85,7 @@ class TestCheckAndroidqfCommand:
|
||||
def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
|
||||
data_path = tmp_path / "androidqf"
|
||||
data_path.mkdir()
|
||||
(data_path / "adb_host_key.pub").write_bytes(b"QUJDRA== acquisition@host\n")
|
||||
(data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
|
||||
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
|
||||
|
||||
command.init()
|
||||
|
||||
@@ -116,7 +116,7 @@ def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
|
||||
module = SysdiagnoseExtraction()
|
||||
command.module_init(module)
|
||||
assert module.tar is None
|
||||
assert Path(module.parent_path) == extracted_path.parent
|
||||
assert module.parent_path == str(extracted_path.parent)
|
||||
finally:
|
||||
command.finish()
|
||||
|
||||
|
||||
@@ -43,7 +43,6 @@ class TestCompletionCommand:
|
||||
|
||||
def test_completion_install_updates_bashrc_once(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("USERPROFILE", str(tmp_path))
|
||||
runner = CliRunner()
|
||||
|
||||
result = runner.invoke(mvt_cli, ["completion", "bash", "--install"])
|
||||
@@ -68,7 +67,6 @@ class TestCompletionCommand:
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("USERPROFILE", str(tmp_path))
|
||||
runner = CliRunner()
|
||||
|
||||
result = runner.invoke(mvt_cli, ["completion", "fish", "--install"])
|
||||
|
||||
Reference in new issue
Block a user