Merge branch 'integrate/hashview-routes'

This commit is contained in:
Justin Bollinger
2026-06-25 11:53:10 -04:00
9 changed files with 771 additions and 119 deletions
+30
View File
@@ -588,6 +588,31 @@ environment variable and provide valid credentials in `config.json`:
HATE_CRACK_RUN_LIVE_TESTS=1 uv run pytest tests/test_upload_cracked_hashes.py -v HATE_CRACK_RUN_LIVE_TESTS=1 uv run pytest tests/test_upload_cracked_hashes.py -v
``` ```
### Live Hashview Tests Against a Local Docker Stack
Instead of pointing the live tests at a remote Hashview server, you can have
the suite spin up a local [Hashview](https://github.com/hashview/hashview)
Docker stack, seed it, run the live tests against it, and tear it down. Set
`HASHVIEW_TEST_LOCAL=1` and point `HASHVIEW_REPO` at a Hashview checkout:
```bash
HASHVIEW_TEST_LOCAL=1 HASHVIEW_REPO=~/projects/hashview \
HATE_CRACK_SKIP_INIT=1 uv run pytest tests/test_hashview_cli_subcommands_subprocess.py -v
```
This brings up `docker compose` in the Hashview repo, seeds an admin API key,
a customer, a hashfile, and cracked "effective task" data, then exports the
`HASHVIEW_*` env vars the tests read. Useful env vars:
- `HASHVIEW_TEST_LOCAL=1` — enable the local stack (no-op otherwise)
- `HASHVIEW_REPO=<path>` — Hashview checkout (default `~/projects/hashview`)
- `HASHVIEW_KEEP=1` — leave containers running after the session (faster re-runs)
- `HASHVIEW_LOCAL_PORT=5000` — host port the app is published on
The hate_crack CLI honours the `HASHVIEW_URL` / `HASHVIEW_API_KEY` environment
variables (overriding `config.json`), which is what lets the suite point the
CLI at the local stack without editing your persisted config.
### End-to-End Install Tests (Local + Docker) ### End-to-End Install Tests (Local + Docker)
Local uv tool install + script execution (uses a temporary HOME): Local uv tool install + script execution (uses a temporary HOME):
@@ -912,6 +937,11 @@ Interactive menu for downloading and managing wordlists from Weakpass.com via Bi
------------------------------------------------------------------- -------------------------------------------------------------------
### Version History ### Version History
Version 2.10.6
- Fixed the Hashview integration calling API routes that don't exist in Hashview (verified against v0.8.3-dev), which 404'd as soon as a customer ID was entered ("Could not list hashfiles"). The customer→hashfile listing relied on a phantom `GET /v1/hashfiles` list-all route; it now enumerates via the real `GET /v1/hashfiles/hash_type/<type>` endpoint where available — the download flow sweeps common hashcat modes to display a customer's uploaded hashfiles. That listing route only exists on Hashview builds from 2026-06-08+ (the `v0.8.3-dev` branch); on `main`/older servers there is no hashfile-listing API at all, so the flow now degrades gracefully to entering the hashfile ID directly (looked up in the Hashview web UI) and resolving its type via `GET /v1/getHashType/<id>`. Additional client-side route fixes: hashfile hash-type lookup now uses `GET /v1/getHashType/<id>`; "left" (uncracked) hash download uses `GET /v1/hashfiles/<id>`; `delete_job` uses `DELETE /v1/jobs/<id>`; `start_job` uses `POST`. Hashview exposes no stop-job route, so `stop_job` now raises with guidance to use `delete_job`; and no bulk cracked-hash export exists, so the best-effort "found" merge degrades gracefully
- The Hashview CLI now honours `HASHVIEW_URL` / `HASHVIEW_API_KEY` environment variables as overrides for the `config.json` values, so the client can be pointed at a different Hashview instance (e.g. a local dev stack) without editing the persisted config
- Added an opt-in local Hashview integration-test harness: `HASHVIEW_TEST_LOCAL=1` (with `HASHVIEW_REPO=<path>`) spins up and seeds a local Hashview docker stack, runs the live Hashview tests against it, and tears it down (`HASHVIEW_KEEP=1` keeps it). This is what surfaced and verified the route fixes above against `v0.8.3-dev`. See the README testing section for details
Version 2.10.5 Version 2.10.5
- Pipal analysis no longer corrupts its input when cracked passwords contain `$HEX[...]` rows. `binascii.unhexlify().decode()` returned the bytes without the trailing newline that normal rows inherit from `password[-1]`, so every HEX-encoded password got concatenated with the next one in the `.passwords` file fed to pipal (e.g. three cracks → two lines, one of them a bogus mashup). Pipal then under-counted entries and reported wrong top base words. The HEX branch now re-appends `\n` so each cracked password lands on its own line - Pipal analysis no longer corrupts its input when cracked passwords contain `$HEX[...]` rows. `binascii.unhexlify().decode()` returned the bytes without the trailing newline that normal rows inherit from `password[-1]`, so every HEX-encoded password got concatenated with the next one in the `.passwords` file fed to pipal (e.g. three cracks → two lines, one of them a bogus mashup). Pipal then under-counted entries and reported wrong top base words. The HEX branch now re-appends `\n` so each cracked password lands on its own line
+143 -58
View File
@@ -1127,7 +1127,7 @@ class HashviewAPI:
def get_hashfile_details(self, hashfile_id): def get_hashfile_details(self, hashfile_id):
"""Get hashfile details and hashtype for a given hashfile_id.""" """Get hashfile details and hashtype for a given hashfile_id."""
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/hash_type" url = f"{self.base_url}/v1/getHashType/{hashfile_id}"
resp = self.session.get(url, headers=self._auth_headers()) resp = self.session.get(url, headers=self._auth_headers())
resp.raise_for_status() resp.raise_for_status()
try: try:
@@ -1173,61 +1173,130 @@ class HashviewAPI:
return {"customers": customers} return {"customers": customers}
return data return data
def list_hashfiles(self): def get_customer_hashfiles(self, customer_id, hash_type=None):
url = f"{self.base_url}/v1/hashfiles" """Return a customer's hashfiles of a given hash_type.
resp = self.session.get(url, headers=self._auth_headers())
resp.raise_for_status()
data = resp.json()
if "hashfiles" in data:
if isinstance(data["hashfiles"], str):
hashfiles = json.loads(data["hashfiles"])
else:
hashfiles = data["hashfiles"]
return hashfiles
return []
def get_customer_hashfiles(self, customer_id): Hashview exposes no "list all hashfiles" route; the only enumeration
all_hashfiles = self.list_hashfiles() endpoint is ``/v1/hashfiles/hash_type/<hash_type>`` (see
:meth:`get_hashfiles_by_type`), which already returns ``customer_id``
and ``hash_type`` per file. We query that and filter by customer.
``hash_type`` is required to enumerate: without it there is no API
route to list a customer's files, so an empty list is returned.
"""
if hash_type is None:
if self.debug:
print(
"[DEBUG] get_customer_hashfiles: no hash_type given; Hashview "
"has no list-all route, returning []"
)
return []
all_hashfiles = self.get_hashfiles_by_type(hash_type)
customer_hfs = [ customer_hfs = [
hf for hf in all_hashfiles if int(hf.get("customer_id", 0)) == customer_id hf for hf in all_hashfiles if int(hf.get("customer_id", 0)) == int(customer_id)
] ]
# The type-scoped endpoint already returns the hash_type, but normalize
# the key so downstream callers can read either spelling.
for hf in customer_hfs:
if not (hf.get("hashtype") or hf.get("hash_type")):
hf["hash_type"] = str(hash_type)
if self.debug: if self.debug:
print( print(
f"[DEBUG] get_customer_hashfiles({customer_id}): found {len(customer_hfs)} hashfiles" f"[DEBUG] get_customer_hashfiles({customer_id}, hash_type={hash_type}): "
f"found {len(customer_hfs)} hashfiles"
) )
# Fetch hash types for any hashfiles missing them
for hf in customer_hfs:
if not (hf.get("hashtype") or hf.get("hash_type")):
hf_id = hf.get("id")
if hf_id is not None:
if self.debug:
print(f"[DEBUG] Fetching hash_type for hashfile {hf_id}")
try:
details = self.get_hashfile_details(hf_id)
hashtype = details.get("hashtype")
if hashtype:
hf["hash_type"] = hashtype
if self.debug:
print(
f"[DEBUG] Updated hashfile {hf_id} with hash_type={hashtype}"
)
elif self.debug:
print(
f"[DEBUG] No hashtype found in details for {hf_id}: {details}"
)
except Exception as e:
if self.debug:
print(
f"[DEBUG] Exception fetching hash_type for {hf_id}: {e}"
)
return customer_hfs return customer_hfs
# Curated set of hashcat modes commonly seen in engagements. Used to
# enumerate a customer's hashfiles by sweeping the per-type listing
# endpoint, since Hashview exposes no list-all route. Not exhaustive:
# uncommon types can still be queried explicitly via get_customer_hashfiles.
COMMON_HASH_TYPES = (
0, # MD5
100, # SHA1
1000, # NTLM
3000, # LM
1100, # Domain Cached Credentials (DCC), MS Cache
2100, # Domain Cached Credentials 2 (DCC2), MS Cache 2
5500, # NetNTLMv1
5600, # NetNTLMv2
7500, # Kerberos 5 AS-REQ Pre-Auth (etype 23)
13100, # Kerberos 5 TGS-REP (Kerberoasting, etype 23)
18200, # Kerberos 5 AS-REP (AS-REP roasting, etype 23)
19600, # Kerberos 5 TGS-REP (etype 17)
19700, # Kerberos 5 TGS-REP (etype 18)
1800, # sha512crypt $6$ (Linux)
500, # md5crypt $1$ (Linux/Cisco)
7400, # sha256crypt $5$
3200, # bcrypt $2*$
1700, # SHA512
1400, # SHA256
160, # HMAC-SHA1
13400, # KeePass
9600, # MS Office 2013
10500, # PDF 1.4-1.6
11600, # 7-Zip
16500, # JWT
22000, # WPA-PBKDF2-PMKID+EAPOL
)
def get_all_customer_hashfiles(self, customer_id, hash_types=None):
"""Aggregate a customer's hashfiles across hash types (deduped by id).
Hashview has no list-all route, so this sweeps the per-type listing
endpoint (:meth:`get_hashfiles_by_type`) over ``hash_types`` and keeps
the files belonging to ``customer_id``. Defaults to
:attr:`COMMON_HASH_TYPES`; a file of an uncommon type not in the sweep
won't appear. The first hash type a file is seen under is retained as
its ``hash_type`` (mixed-type files are rare).
"""
if hash_types is None:
hash_types = self.COMMON_HASH_TYPES
seen = {}
for ht in hash_types:
try:
files = self.get_hashfiles_by_type(ht)
except Exception as e:
status = getattr(getattr(e, "response", None), "status_code", None)
if status == 404:
# The /v1/hashfiles/hash_type route doesn't exist on this
# server (e.g. Hashview main, or builds before 2026-06-08),
# so no per-type sweep is possible. Stop after the first 404
# instead of hammering the server with one request per type.
if self.debug:
print(
"[DEBUG] get_all_customer_hashfiles: hash_type listing "
"endpoint absent (404); aborting sweep"
)
break
if self.debug:
print(f"[DEBUG] get_all_customer_hashfiles: type {ht} failed: {e}")
continue
for hf in files:
if int(hf.get("customer_id", 0)) != int(customer_id):
continue
hf_id = hf.get("id")
if hf_id is None:
continue
if not (hf.get("hashtype") or hf.get("hash_type")):
hf["hash_type"] = str(ht)
seen.setdefault(int(hf_id), hf)
if self.debug:
print(
f"[DEBUG] get_all_customer_hashfiles({customer_id}): "
f"found {len(seen)} hashfiles across {len(hash_types)} types"
)
return list(seen.values())
def get_customer_hashfiles_with_hashtype(self, customer_id, target_hashtype="1000"): def get_customer_hashfiles_with_hashtype(self, customer_id, target_hashtype="1000"):
"""Return hashfiles for a customer that match the requested hashtype.""" """Return hashfiles for a customer that match the requested hashtype."""
customer_hashfiles = self.get_customer_hashfiles(customer_id) customer_hashfiles = self.get_customer_hashfiles(
customer_id, hash_type=target_hashtype
)
if not customer_hashfiles: if not customer_hashfiles:
return [] return []
target_str = str(target_hashtype) target_str = str(target_hashtype)
@@ -1316,26 +1385,29 @@ class HashviewAPI:
return {} return {}
def stop_job(self, job_id): def stop_job(self, job_id):
url = f"{self.base_url}/v1/jobs/stop/{job_id}" # Hashview exposes no "stop job" route (only add/get/start/delete).
resp = self.session.get(url) # Deleting a job removes it regardless of Queued/Running state, which
resp.raise_for_status() # is the closest supported operation; use delete_job() instead.
return resp.json() raise NotImplementedError(
"Hashview has no stop-job endpoint; use delete_job() to remove a job."
)
def delete_job(self, job_id): def delete_job(self, job_id):
url = f"{self.base_url}/v1/jobs/delete/{job_id}" # Hashview deletes via DELETE /v1/jobs/<id> (there is no /jobs/delete/).
resp = self.session.get(url) url = f"{self.base_url}/v1/jobs/{job_id}"
resp = self.session.delete(url)
resp.raise_for_status() resp.raise_for_status()
return resp.json() return resp.json()
def start_job(self, job_id, priority=3, limit_recovered=False): def start_job(self, job_id, priority=3, limit_recovered=False):
# /v1/jobs/start/<id> is POST-only; priority/limit_recovered come from
# the stored job record server-side, so they are validated here but not
# required by the endpoint.
url = f"{self.base_url}/v1/jobs/start/{job_id}" url = f"{self.base_url}/v1/jobs/start/{job_id}"
params = {}
priority = int(priority) priority = int(priority)
if priority < 1 or priority > 5: if priority < 1 or priority > 5:
raise ValueError("priority must be an int between 1 and 5") raise ValueError("priority must be an int between 1 and 5")
params["priority"] = priority resp = self.session.post(url)
params["limit_recovered"] = bool(limit_recovered)
resp = self.session.get(url, params=params)
resp.raise_for_status() resp.raise_for_status()
return resp.json() return resp.json()
@@ -1344,7 +1416,10 @@ class HashviewAPI:
): ):
import sys import sys
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/left" # Hashview's GET /v1/hashfiles/<id> returns exactly the uncracked
# ("left") ciphertexts for the hashfile (see v1_api_get_hashfile). The
# older /v1/hashfiles/<id>/left route no longer exists and 404s.
url = f"{self.base_url}/v1/hashfiles/{hashfile_id}"
resp = self.session.get(url, headers=self._auth_headers(), stream=True) resp = self.session.get(url, headers=self._auth_headers(), stream=True)
resp.raise_for_status() resp.raise_for_status()
if output_file is None: if output_file is None:
@@ -1383,7 +1458,10 @@ class HashviewAPI:
found_file = os.path.join(out_dir, f"found_{customer_id}_{hashfile_id}.txt") found_file = os.path.join(out_dir, f"found_{customer_id}_{hashfile_id}.txt")
try: try:
# Try to download the found file # Best-effort: Hashview v0.8.3-dev exposes no bulk "found"/cracked
# export endpoint (only the single-hash POST /v1/search), so this
# request 404s against stock servers and the merge is skipped. It
# remains for forks/versions that expose a per-hashfile found dump.
found_url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/found" found_url = f"{self.base_url}/v1/hashfiles/{hashfile_id}/found"
found_resp = self.session.get( found_resp = self.session.get(
found_url, headers=self._auth_headers(), stream=True, timeout=30 found_url, headers=self._auth_headers(), stream=True, timeout=30
@@ -1604,8 +1682,13 @@ def download_hashes_from_hashview(
input_fn: Callable[[str], str] = input, input_fn: Callable[[str], str] = input,
print_fn: Callable[..., None] = print, print_fn: Callable[..., None] = print,
potfile_path: Optional[str] = None, potfile_path: Optional[str] = None,
hash_type: Optional[str] = None,
) -> Tuple[str, str]: ) -> Tuple[str, str]:
"""Interactive Hashview download flow used by CLI.""" """Interactive Hashview download flow used by CLI.
``hash_type`` is required to enumerate a customer's hashfiles, since
Hashview only exposes a per-hash-type listing endpoint.
"""
try: try:
if not sys.stdin or not sys.stdin.isatty(): if not sys.stdin or not sys.stdin.isatty():
print_fn("\nAvailable Customers:") print_fn("\nAvailable Customers:")
@@ -1663,7 +1746,9 @@ def download_hashes_from_hashview(
else: else:
customer_id = int(customer_raw) customer_id = int(customer_raw)
try: try:
customer_hashfiles = api_harness.get_customer_hashfiles(customer_id) customer_hashfiles = api_harness.get_customer_hashfiles(
customer_id, hash_type=hash_type
)
if customer_hashfiles: if customer_hashfiles:
print_fn("\n" + "=" * 120) print_fn("\n" + "=" * 120)
print_fn(f"Hashfiles for Customer ID {customer_id}:") print_fn(f"Hashfiles for Customer ID {customer_id}:")
+48 -19
View File
@@ -272,8 +272,14 @@ if _missing_keys:
print(f"[config] Added {len(_missing_keys)} missing key(s) to {_config_path}") print(f"[config] Added {len(_missing_keys)} missing key(s) to {_config_path}")
print(f" Keys: {', '.join(_missing_keys)}") print(f" Keys: {', '.join(_missing_keys)}")
hashview_url = config_parser["hashview_url"] # Environment variables override config.json so the CLI can be pointed at a
hashview_api_key = config_parser["hashview_api_key"] # different Hashview instance (e.g. a local docker stack for the live test
# suite) without editing the persisted config. Empty/unset env vars fall back
# to config.json.
hashview_url = os.environ.get("HASHVIEW_URL") or config_parser["hashview_url"]
hashview_api_key = (
os.environ.get("HASHVIEW_API_KEY") or config_parser["hashview_api_key"]
)
SKIP_INIT = os.environ.get("HATE_CRACK_SKIP_INIT") == "1" SKIP_INIT = os.environ.get("HATE_CRACK_SKIP_INIT") == "1"
@@ -3710,6 +3716,7 @@ def hashview_api():
elif option_key == "download_hashes": elif option_key == "download_hashes":
# Download left hashes # Download left hashes
try: try:
cancel_download = False
while True: while True:
# First, list customers to help user select # First, list customers to help user select
customers_result = api_harness.list_customers() customers_result = api_harness.list_customers()
@@ -3757,24 +3764,31 @@ def hashview_api():
) )
continue continue
# List hashfiles for the customer # Try to list the customer's hashfiles for convenience.
# This only works on Hashview servers that expose the
# /v1/hashfiles/hash_type endpoint (v0.8.3-dev, added
# 2026-06-08); on `main` or older builds there is no
# hashfile-listing API, so we fall back to entering the
# hashfile ID directly (look it up in the web UI).
hashfile_map = {}
try: try:
customer_hashfiles = api_harness.get_customer_hashfiles( print(
customer_id "\nScanning customer hashfiles across common hash types..."
) )
customer_hashfiles = (
api_harness.get_all_customer_hashfiles(customer_id)
)
except Exception as e:
customer_hashfiles = []
if debug_mode:
print(f"[DEBUG] hashfile listing unavailable: {e}")
if not customer_hashfiles: if customer_hashfiles:
print(
f"\nNo hashfiles found for customer ID {customer_id}"
)
continue
print("\n" + "=" * 120) print("\n" + "=" * 120)
print(f"Hashfiles for Customer ID {customer_id}:") print(f"Hashfiles for Customer ID {customer_id}:")
print("=" * 120) print("=" * 120)
print(f"{'ID':<10} {'Hash Type':<10} {'Name':<96}") print(f"{'ID':<10} {'Hash Type':<10} {'Name':<96}")
print("-" * 120) print("-" * 120)
hashfile_map = {}
for hf in customer_hashfiles: for hf in customer_hashfiles:
hf_id = hf.get("id") hf_id = hf.get("id")
hf_name = hf.get("name", "N/A") hf_name = hf.get("name", "N/A")
@@ -3794,22 +3808,33 @@ def hashview_api():
hashfile_map[int(hf_id)] = hf_type hashfile_map[int(hf_id)] = hf_type
print("=" * 120) print("=" * 120)
print(f"Total: {len(hashfile_map)} hashfile(s)") print(f"Total: {len(hashfile_map)} hashfile(s)")
except Exception as e: else:
print(f"\nWarning: Could not list hashfiles: {e}") print(
continue "\nThis Hashview server has no hashfile-listing API "
"(it lacks the /v1/hashfiles/hash_type endpoint), or "
f"customer {customer_id} has no hashfiles of a common "
"type. Look up the hashfile ID in the Hashview web UI "
"and enter it below."
)
while True: while True:
hashfile_id_input = input(
"\nEnter hashfile ID (or Q to cancel): "
).strip()
if hashfile_id_input.lower() == "q":
cancel_download = True
break
try: try:
hashfile_id_input = input(
"\nEnter hashfile ID: "
).strip()
hashfile_id = int(hashfile_id_input) hashfile_id = int(hashfile_id_input)
except ValueError: except ValueError:
print( print(
"\n✗ Error: Invalid ID entered. Please enter a numeric ID." "\n✗ Error: Invalid ID entered. Please enter a numeric ID."
) )
continue continue
if hashfile_id not in hashfile_map: # Only restrict to the listed set when we actually
# have a listing; otherwise accept any ID the user
# read from the web UI.
if hashfile_map and hashfile_id not in hashfile_map:
print( print(
"\n✗ Error: Hashfile ID not in the list. Please try again." "\n✗ Error: Hashfile ID not in the list. Please try again."
) )
@@ -3817,6 +3842,10 @@ def hashview_api():
break break
break break
# User cancelled at the hash-type prompt: back to the menu.
if cancel_download:
continue
# Set output filename automatically # Set output filename automatically
output_file = f"left_{customer_id}_{hashfile_id}.txt" output_file = f"left_{customer_id}_{hashfile_id}.txt"
+231
View File
@@ -0,0 +1,231 @@
"""Bring up / tear down a local Hashview docker stack for the live test suite.
Used by the session-scoped ``hashview_local_stack`` fixture in ``conftest.py``.
Activated by ``HASHVIEW_TEST_LOCAL=1``; otherwise everything here is a no-op and
the live tests skip as before.
Lifecycle (when enabled):
1. write ``hashview/config.conf`` with ``SERVER_NAME = 127.0.0.1:5000`` so the
app is reachable from the host (a stale ``app:5000`` SERVER_NAME makes every
request 404 via Flask host matching)
2. ``docker compose up -d --build``
3. poll until the app answers on the host
4. seed the DB (admin api_key, customer, hashfile, cracked effective-task data)
5. export ``HASHVIEW_*`` env vars so both the test helpers and the hate_crack
CLI (which now honours ``HASHVIEW_URL`` / ``HASHVIEW_API_KEY``) target local
6. on teardown ``docker compose down -v`` unless ``HASHVIEW_KEEP=1``
Config via env:
HASHVIEW_TEST_LOCAL=1 enable the stack
HASHVIEW_REPO=<path> hashview checkout (default ~/projects/hashview)
HASHVIEW_KEEP=1 leave containers running after the session
HASHVIEW_LOCAL_PORT=5000 host port the app is published on
"""
import os
import shutil
import subprocess
import time
import urllib.error
import urllib.request
from pathlib import Path
LOCAL_API_KEY = "3eac1ab7-e525-4bb7-b565-2c9f045dfc56"
CUSTOMER_ID = "1"
HASHFILE_ID = "1"
# md5 is hashcat mode 0; the suite's pwdump test uploads NTLM (1000). Seed both.
HASH_TYPE = "0"
SEED_HASH_TYPES = "0,1000"
_CONFIG_CONF = """[SERVER]
SERVER_NAME = 127.0.0.1:{port}
SECRET_KEY = hate-crack-local-test-secret
[database]
host = db
username = hashview
password = hashview
[SMTP]
server = smtp.example.com
port = 25
use_tls = False
username =
password =
default_sender =
"""
def enabled() -> bool:
return os.environ.get("HASHVIEW_TEST_LOCAL", "").lower() in ("1", "true", "yes")
def _keep() -> bool:
return os.environ.get("HASHVIEW_KEEP", "").lower() in ("1", "true", "yes")
def _repo() -> Path:
return Path(
os.environ.get("HASHVIEW_REPO", os.path.expanduser("~/projects/hashview"))
).resolve()
def _port() -> str:
return os.environ.get("HASHVIEW_LOCAL_PORT", "5000")
def _base_url() -> str:
return f"http://127.0.0.1:{_port()}"
def _compose(repo: Path, *args: str, check: bool = True, capture: bool = False):
env = {**os.environ, "DOCKER_PLATFORM": os.environ.get("DOCKER_PLATFORM", "linux/amd64")}
return subprocess.run(
["docker", "compose", *args],
cwd=str(repo),
env=env,
check=check,
text=True,
stdout=subprocess.PIPE if capture else None,
stderr=subprocess.STDOUT if capture else None,
)
def _http_status(url: str, cookie: str | None = None):
"""Return the HTTP status for ``url`` (following no redirects), or None."""
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs):
return None
opener = urllib.request.build_opener(_NoRedirect)
req = urllib.request.Request(url)
if cookie:
req.add_header("Cookie", f"uuid={cookie}")
try:
with opener.open(req, timeout=3) as resp:
return resp.status
except urllib.error.HTTPError as exc:
return exc.code
except Exception:
return None
# Statuses that mean "Flask is up and routing" — i.e. SERVER_NAME matches and
# the app is serving. ``/login`` answers 200 or 302 (redirect) depending on
# session/setup state; either is fine. A 404 means the route didn't match
# (e.g. a stale ``app:5000`` SERVER_NAME) and a connection error (None) means
# the app isn't listening yet.
_ROUTING_STATUSES = {200, 301, 302, 303, 403}
def _wait_ready(timeout: float = 240.0) -> bool:
"""Poll until the app is routing requests on ``/login``.
On a cold start the DB volume is fresh, so the app boots and runs
migrations before it stops connection-refusing. Routing (not auth/seed)
readiness is the gate here; auth/seed readiness is checked separately in
:func:`_wait_authenticated`.
"""
url = f"{_base_url()}/login"
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if _http_status(url) in _ROUTING_STATUSES:
return True
time.sleep(2)
return False
def _wait_authenticated(timeout: float = 60.0) -> bool:
"""Poll an authenticated endpoint until the seeded api_key is accepted.
Closes the race where ``/login`` is up but the seed's admin api_key isn't
effective yet: an unauthenticated ``/v1/customers`` 302-redirects (to an
HTML page), while a recognised cookie returns 200 JSON.
"""
url = f"{_base_url()}/v1/customers"
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if _http_status(url, cookie=LOCAL_API_KEY) == 200:
return True
time.sleep(2)
return False
def _seed(repo: Path, attempts: int = 6) -> None:
seeder = Path(__file__).with_name("hashview_local_seed.py")
_compose(repo, "cp", str(seeder), "app:/tmp/hashview_local_seed.py")
seed_env = [
"-e", "PYTHONPATH=/",
"-e", f"HASHVIEW_API_KEY={LOCAL_API_KEY}",
"-e", f"HASHVIEW_CUSTOMER_ID={CUSTOMER_ID}",
"-e", f"HASHVIEW_HASHFILE_ID={HASHFILE_ID}",
"-e", f"HASHVIEW_SEED_HASH_TYPES={SEED_HASH_TYPES}",
]
# The admin user (id=1) and default tasks are created during the app's
# first-request boot, which can lag a few seconds behind the app routing.
# Retry so a cold start doesn't fail on "Admin user not found".
last = None
for i in range(attempts):
result = _compose(
repo,
"exec",
"-T",
*seed_env,
"-w",
"/",
"app",
"python",
"/tmp/hashview_local_seed.py",
check=False,
capture=True,
)
if result.returncode == 0:
return
last = result.stdout
time.sleep(5)
raise RuntimeError(f"seed did not succeed after {attempts} attempts: {last}")
def _export_env() -> None:
os.environ["HASHVIEW_TEST_REAL"] = "1"
os.environ["HASHVIEW_URL"] = _base_url()
os.environ["HASHVIEW_API_KEY"] = LOCAL_API_KEY
os.environ["HASHVIEW_CUSTOMER_ID"] = CUSTOMER_ID
os.environ["HASHVIEW_HASHFILE_ID"] = HASHFILE_ID
os.environ.setdefault("HASHVIEW_HASH_TYPE", HASH_TYPE)
def setup():
"""Bring the stack up + seed. Returns a skip reason string, or None on success."""
if shutil.which("docker") is None:
return "docker not available"
repo = _repo()
if not repo.is_dir():
return f"hashview repo not found at {repo} (set HASHVIEW_REPO)"
if not (repo / "docker-compose.yml").is_file():
return f"no docker-compose.yml in hashview repo {repo}"
(repo / "hashview" / "config.conf").write_text(_CONFIG_CONF.format(port=_port()))
try:
_compose(repo, "up", "-d", "--build")
except subprocess.CalledProcessError as exc:
return f"docker compose up failed: {exc}"
if not _wait_ready():
return f"hashview app did not become ready at {_base_url()}"
try:
_seed(repo)
except (subprocess.CalledProcessError, RuntimeError) as exc:
return f"hashview DB seed failed: {exc}"
if not _wait_authenticated():
return "seeded api_key not accepted by Hashview (auth/seed race)"
_export_env()
return None
def teardown():
if _keep():
return
repo = _repo()
if repo.is_dir() and (repo / "docker-compose.yml").is_file():
_compose(repo, "down", "-v", check=False, capture=True)
+37
View File
@@ -42,3 +42,40 @@ def _isolate_notify_state():
notify.clear_state_for_tests() notify.clear_state_for_tests()
yield yield
notify.clear_state_for_tests() notify.clear_state_for_tests()
def pytest_configure(config):
"""Spin up + seed a local Hashview docker stack for the live test suite.
No-op unless ``HASHVIEW_TEST_LOCAL=1``. When enabled, brings up the stack
from the hashview repo, seeds the DB, and exports the ``HASHVIEW_*`` env
vars the live tests (and the hate_crack CLI) read so they target the local
instance instead of whatever ``config.json`` points at.
This runs in ``pytest_configure`` *before* collection on purpose: the
live subprocess tests gate on ``HASHVIEW_TEST_REAL`` via ``@skipif``, which
is evaluated at collection time. A session fixture would set the env too
late and every live test would skip. On failure we deliberately leave
``HASHVIEW_TEST_REAL`` unset so the live tests skip with their normal
reason rather than erroring. See ``tests/_hashview_local.py`` for config.
"""
from tests import _hashview_local as hv
if not hv.enabled():
return
reason = hv.setup()
if reason is not None:
config.issue_config_time_warning(
pytest.PytestWarning(
f"HASHVIEW_TEST_LOCAL set but local stack unavailable "
f"({reason}); live Hashview tests will skip."
),
stacklevel=2,
)
def pytest_unconfigure(config):
from tests import _hashview_local as hv
if hv.enabled():
hv.teardown()
+152
View File
@@ -0,0 +1,152 @@
"""Seed a local Hashview docker stack to the state hate_crack's live tests need.
Runs INSIDE the Hashview ``app`` container (``PYTHONPATH=/``), where the
``hashview`` package is importable. Idempotent.
Brings the database to:
- admin user (id=1) with a known ``api_key`` (so hate_crack can authenticate)
- a Settings row (so the /setup wizard does not intercept requests)
- a Customer at ``HASHVIEW_CUSTOMER_ID``
- a Hashfile at ``HASHVIEW_HASHFILE_ID`` holding one (uncracked) hash, so the
``download-hashes`` / left-list endpoints return data
- at least one *cracked* Hashes row per hash type in
``HASHVIEW_SEED_HASH_TYPES`` (comma separated, default ``0,1000``) attached to
a real Task. ``/v1/jobs/add`` derives its "top 10 effective tasks" from
cracked hashes of the uploaded file's hash type; without this it answers
"Not enough data to determine effective tasks" and job creation fails.
Required env: HASHVIEW_API_KEY, HASHVIEW_CUSTOMER_ID, HASHVIEW_HASHFILE_ID.
Optional env: HASHVIEW_SEED_HASH_TYPES, HASHVIEW_SEED_TASK_ID (default 1).
"""
import os
import sys
from flask import Flask
from flask_bcrypt import Bcrypt
from hashview.config import Config
from hashview.models import (
Customers,
Hashes,
HashfileHashes,
Hashfiles,
Settings,
Tasks,
Users,
db,
)
REQUIRED_ENV = (
"HASHVIEW_API_KEY",
"HASHVIEW_CUSTOMER_ID",
"HASHVIEW_HASHFILE_ID",
)
def build_app() -> Flask:
app = Flask(__name__)
app.config.from_object(Config)
db.init_app(app)
Bcrypt(app)
return app
def seed(app: Flask) -> None:
api_key = os.environ["HASHVIEW_API_KEY"]
customer_id = int(os.environ["HASHVIEW_CUSTOMER_ID"])
hashfile_id = int(os.environ["HASHVIEW_HASHFILE_ID"])
task_id = int(os.environ.get("HASHVIEW_SEED_TASK_ID", "1"))
hash_types = [
int(x) for x in os.environ.get("HASHVIEW_SEED_HASH_TYPES", "0,1000").split(",") if x.strip()
]
bcrypt = Bcrypt(app)
with app.app_context():
admin = db.session.get(Users, 1)
if admin is None:
raise RuntimeError(
"Admin user (id=1) not found. App must finish booting before seeding."
)
admin.api_key = api_key
admin.admin = True
# Hashview's setup guard (do_gui_setup_if_needed) redirects every request
# to /setup/admin-pass while the admin password is still the default.
# Set a non-default password so the API is reachable without the wizard.
admin.password = bcrypt.generate_password_hash(
os.environ.get("HASHVIEW_ADMIN_PASSWORD", "hate-crack-local-pass")
).decode("utf-8")
admin.email_address = admin.email_address or "admin@hate-crack.local"
if db.session.query(Settings).first() is None:
db.session.add(
Settings(retention_period=30, max_runtime_jobs=0, max_runtime_tasks=0)
)
if db.session.get(Customers, customer_id) is None:
db.session.add(Customers(id=customer_id, name="hate_crack Local Customer"))
if db.session.get(Tasks, task_id) is None:
raise RuntimeError(
f"Task id={task_id} not found. Default tasks should exist after app boot."
)
if db.session.get(Hashfiles, hashfile_id) is None:
db.session.add(
Hashfiles(
id=hashfile_id,
name="hate-crack-local-hashfile",
customer_id=customer_id,
owner_id=admin.id,
)
)
db.session.flush()
hash_row = Hashes(
sub_ciphertext="d41d8cd98f00b204e9800998ecf8427e",
ciphertext="d41d8cd98f00b204e9800998ecf8427e",
hash_type=0,
cracked=False,
)
db.session.add(hash_row)
db.session.flush()
db.session.add(
HashfileHashes(
hash_id=hash_row.id, username="local-user", hashfile_id=hashfile_id
)
)
# Cracked "effective task" data, one row per hash type the tests upload.
for ht in hash_types:
sub = f"seed-cracked-{ht}"
exists = (
db.session.query(Hashes)
.filter_by(sub_ciphertext=sub, hash_type=ht, cracked=True)
.first()
)
if exists is None:
db.session.add(
Hashes(
sub_ciphertext=sub,
ciphertext=sub,
hash_type=ht,
cracked=True,
task_id=task_id,
plaintext="password",
)
)
db.session.commit()
def main() -> int:
missing = [k for k in REQUIRED_ENV if not os.getenv(k)]
if missing:
print(f"hashview_local_seed: missing env vars: {missing}", file=sys.stderr)
return 2
seed(build_app())
print("hashview_local_seed: ok")
return 0
if __name__ == "__main__":
sys.exit(main())
+106 -37
View File
@@ -68,74 +68,107 @@ class TestHashviewAPI:
# Cleanup # Cleanup
def test_list_hashfiles_success(self, api): def test_get_hashfiles_by_type_success(self, api):
"""Test successful hashfile listing with real API if possible, else mock.""" """The /v1/hashfiles/hash_type/<type> endpoint returns a list (real API if possible)."""
hashview_url, hashview_api_key = self._get_hashview_config() hashview_url, hashview_api_key = self._get_hashview_config()
if hashview_url and hashview_api_key: if hashview_url and hashview_api_key:
real_api = HashviewAPI(hashview_url, hashview_api_key) real_api = HashviewAPI(hashview_url, hashview_api_key)
result = real_api.list_hashfiles() result = real_api.get_hashfiles_by_type("1000")
assert isinstance(result, list) assert isinstance(result, list)
# If there are no hashfiles, that's valid, but if present, check structure
if result: if result:
assert "name" in result[0] assert "name" in result[0]
else: else:
mock_response = Mock() mock_response = Mock()
mock_response.json.return_value = { mock_response.json.return_value = [
"hashfiles": json.dumps( {"id": 1, "customer_id": 1, "name": "hashfile1.txt", "hash_type": 1000},
[ {"id": 2, "customer_id": 2, "name": "hashfile2.txt", "hash_type": 1000},
{"id": 1, "customer_id": 1, "name": "hashfile1.txt"}, ]
{"id": 2, "customer_id": 2, "name": "hashfile2.txt"},
]
)
}
mock_response.raise_for_status = Mock() mock_response.raise_for_status = Mock()
api.session.get.return_value = mock_response api.session.get.return_value = mock_response
result = api.list_hashfiles() result = api.get_hashfiles_by_type("1000")
assert isinstance(result, list) assert isinstance(result, list)
assert len(result) == 2 assert len(result) == 2
assert result[0]["name"] == "hashfile1.txt" assert result[0]["name"] == "hashfile1.txt"
def test_list_hashfiles_empty(self, api): def test_get_customer_hashfiles_requires_hash_type(self, api):
"""Test hashfile listing returns empty list if no hashfiles (real API if possible).""" """Without a hash_type there is no Hashview list route, so we return []."""
hashview_url, hashview_api_key = self._get_hashview_config() result = api.get_customer_hashfiles(1)
if hashview_url and hashview_api_key: assert result == []
real_api = HashviewAPI(hashview_url, hashview_api_key)
result = real_api.list_hashfiles() def test_get_all_customer_hashfiles_sweeps_and_dedupes(self, api):
# If there are no hashfiles, result should be [] """Aggregate sweeps per-type listings, filters by customer, dedupes by id."""
if not result: per_type = {
assert result == [] 1000: [
else: {"id": 1, "customer_id": 1, "name": "ntlm.txt", "hash_type": 1000},
assert isinstance(result, list) {"id": 2, "customer_id": 2, "name": "other.txt", "hash_type": 1000},
else: ],
mock_response = Mock() 5600: [
mock_response.json.return_value = {} {"id": 3, "customer_id": 1, "name": "ntlmv2.txt", "hash_type": 5600},
mock_response.raise_for_status = Mock() # id 1 appears again under another type; must dedupe (first wins)
api.session.get.return_value = mock_response {"id": 1, "customer_id": 1, "name": "ntlm.txt", "hash_type": 5600},
result = api.list_hashfiles() ],
assert result == [] }
api.get_hashfiles_by_type = Mock(side_effect=lambda ht: per_type.get(int(ht), []))
result = api.get_all_customer_hashfiles(1, hash_types=[1000, 5600])
ids = sorted(hf["id"] for hf in result)
assert ids == [1, 3] # customer 2 excluded, id 1 not duplicated
by_id = {hf["id"]: hf for hf in result}
assert str(by_id[1]["hash_type"]) == "1000" # first type seen wins
assert str(by_id[3]["hash_type"]) == "5600"
def test_get_all_customer_hashfiles_aborts_on_404(self, api):
"""A 404 means the listing endpoint is absent (e.g. Hashview main);
the sweep stops after the first request instead of probing every type."""
import requests
def _raise_404(ht):
resp = Mock()
resp.status_code = 404
raise requests.exceptions.HTTPError("404 Not Found", response=resp)
api.get_hashfiles_by_type = Mock(side_effect=_raise_404)
result = api.get_all_customer_hashfiles(1, hash_types=[1000, 5600, 3000])
assert result == []
# Stopped after the first 404, did not sweep all three types.
assert api.get_hashfiles_by_type.call_count == 1
def test_get_all_customer_hashfiles_skips_failing_types(self, api):
"""A per-type query that errors is skipped, not fatal."""
def _by_type(ht):
if int(ht) == 1000:
raise RuntimeError("boom")
return [{"id": 9, "customer_id": 1, "name": "x", "hash_type": int(ht)}]
api.get_hashfiles_by_type = Mock(side_effect=_by_type)
result = api.get_all_customer_hashfiles(1, hash_types=[1000, 5600])
assert [hf["id"] for hf in result] == [9]
def test_get_customer_hashfiles(self, api): def test_get_customer_hashfiles(self, api):
"""Test filtering hashfiles by customer_id (real API if possible).""" """Filter the type-scoped hashfile list by customer_id (real API if possible)."""
hashview_url, hashview_api_key = self._get_hashview_config() hashview_url, hashview_api_key = self._get_hashview_config()
customer_id = os.environ.get("HASHVIEW_CUSTOMER_ID") customer_id = os.environ.get("HASHVIEW_CUSTOMER_ID")
if hashview_url and hashview_api_key and customer_id: if hashview_url and hashview_api_key and customer_id:
real_api = HashviewAPI(hashview_url, hashview_api_key) real_api = HashviewAPI(hashview_url, hashview_api_key)
result = real_api.get_customer_hashfiles(int(customer_id)) result = real_api.get_customer_hashfiles(int(customer_id), hash_type="1000")
assert isinstance(result, list) assert isinstance(result, list)
# If there are hashfiles, all should match customer_id # If there are hashfiles, all should match customer_id
if result: if result:
assert all(hf["customer_id"] == int(customer_id) for hf in result) assert all(hf["customer_id"] == int(customer_id) for hf in result)
else: else:
api.list_hashfiles = Mock( api.get_hashfiles_by_type = Mock(
return_value=[ return_value=[
{"id": 1, "customer_id": 1, "name": "hashfile1.txt"}, {"id": 1, "customer_id": 1, "name": "hashfile1.txt"},
{"id": 2, "customer_id": 2, "name": "hashfile2.txt"}, {"id": 2, "customer_id": 2, "name": "hashfile2.txt"},
{"id": 3, "customer_id": 1, "name": "hashfile3.txt"}, {"id": 3, "customer_id": 1, "name": "hashfile3.txt"},
] ]
) )
result = api.get_customer_hashfiles(1) result = api.get_customer_hashfiles(1, hash_type="1000")
assert len(result) == 2 assert len(result) == 2
assert all(hf["customer_id"] == 1 for hf in result) assert all(hf["customer_id"] == 1 for hf in result)
api.get_hashfiles_by_type.assert_called_once_with("1000")
def test_display_customers_multicolumn_empty(self, api, capsys): def test_display_customers_multicolumn_empty(self, api, capsys):
"""Test display_customers_multicolumn with no customers (mock only, as real API not needed).""" """Test display_customers_multicolumn with no customers (mock only, as real API not needed)."""
@@ -262,9 +295,15 @@ class TestHashviewAPI:
assert content == b"hash1\nhash2\n" assert content == b"hash1\nhash2\n"
assert result["size"] == len(content) assert result["size"] == len(content)
# Verify auth headers were passed in the left hashes download call # Verify auth headers were passed in the left hashes download call.
# The uncracked ("left") hashes come from GET /v1/hashfiles/<id>
# (the trailing /found call is a separate lookup).
call_args_list = api.session.get.call_args_list call_args_list = api.session.get.call_args_list
left_call = [c for c in call_args_list if "left" in str(c)][0] left_call = [
c
for c in call_args_list
if "/v1/hashfiles/2" in str(c) and "found" not in str(c)
][0]
assert left_call.kwargs.get("headers") is not None assert left_call.kwargs.get("headers") is not None
auth_headers = left_call.kwargs.get("headers") auth_headers = left_call.kwargs.get("headers")
assert "Cookie" in auth_headers or "uuid" in str(auth_headers) assert "Cookie" in auth_headers or "uuid" in str(auth_headers)
@@ -469,6 +508,36 @@ class TestHashviewAPI:
print("✓ Option 2 (Create Job) is READY and WORKING!") print("✓ Option 2 (Create Job) is READY and WORKING!")
print("=" * 60) print("=" * 60)
def test_start_job_uses_post(self, api):
"""start_job must POST to /v1/jobs/start/<id> (the route is POST-only)."""
mock_response = Mock()
mock_response.json.return_value = {"status": 200, "msg": "Job started"}
mock_response.raise_for_status = Mock()
api.session.post.return_value = mock_response
result = api.start_job(42)
assert result["msg"] == "Job started"
api.session.post.assert_called_once_with(f"{HASHVIEW_URL}/v1/jobs/start/42")
api.session.get.assert_not_called()
def test_delete_job_uses_delete_verb(self, api):
"""delete_job must use DELETE /v1/jobs/<id> (there is no /jobs/delete/)."""
mock_response = Mock()
mock_response.json.return_value = {"status": 200, "msg": "Job deleted"}
mock_response.raise_for_status = Mock()
api.session.delete.return_value = mock_response
result = api.delete_job(7)
assert result["msg"] == "Job deleted"
api.session.delete.assert_called_once_with(f"{HASHVIEW_URL}/v1/jobs/7")
def test_stop_job_not_supported(self, api):
"""Hashview has no stop-job route, so stop_job raises NotImplementedError."""
with pytest.raises(NotImplementedError):
api.stop_job(7)
def test_create_job_with_new_customer(self, api, test_hashfile): def test_create_job_with_new_customer(self, api, test_hashfile):
"""Test creating a new customer and then creating a job (real API if possible).""" """Test creating a new customer and then creating a job (real API if possible)."""
hashview_url, hashview_api_key = self._get_hashview_config() hashview_url, hashview_api_key = self._get_hashview_config()
@@ -105,6 +105,16 @@ def test_hashview_subcommands_require_api_key(tmp_path, args):
path.write_text("dummy\n") path.write_text("dummy\n")
args[idx + 1] = str(path) args[idx + 1] = str(path)
# Strip any ambient Hashview credentials (e.g. exported by the local-stack
# fixture) so this exercises the genuine no-key-configured path. The CLI
# honours HASHVIEW_URL / HASHVIEW_API_KEY as overrides, so leaving them set
# would supply a key and defeat the check.
sub_env = {
k: v
for k, v in os.environ.items()
if k not in ("HASHVIEW_URL", "HASHVIEW_API_KEY")
}
sub_env["PYTHONUNBUFFERED"] = "1"
cli_cmd = [sys.executable, HATE_CRACK_SCRIPT] + args cli_cmd = [sys.executable, HATE_CRACK_SCRIPT] + args
result = subprocess.run( result = subprocess.run(
cli_cmd, cli_cmd,
@@ -112,7 +122,7 @@ def test_hashview_subcommands_require_api_key(tmp_path, args):
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, text=True,
cwd=REPO_ROOT, cwd=REPO_ROOT,
env={**os.environ, "PYTHONUNBUFFERED": "1"}, env=sub_env,
) )
output = result.stdout + result.stderr output = result.stdout + result.stderr
assert "Hashview API key not configured" in output assert "Hashview API key not configured" in output
@@ -215,7 +225,10 @@ def test_hashview_subcommands_live_upload_hashfile_job(tmp_path):
) )
assert run.returncode == 0, output assert run.returncode == 0, output
assert ("Hashfile uploaded" in output) or ("Hashfile added" in output) assert ("Hashfile uploaded" in output) or ("Hashfile added" in output)
assert ("Job created" in output) or ("Failed to add job" in output) # Success surfaces the server's job id ("Job ID: N"); a graceful failure
# surfaces an "Error:" line. The CLI echoes the server's own message ("Job
# added"), so don't assert on the literal "Job created".
assert ("Job ID:" in output) or ("Error:" in output)
if "Job ID:" in output: if "Job ID:" in output:
job_id = None job_id = None
for line in output.splitlines(): for line in output.splitlines():
@@ -305,7 +318,10 @@ def test_hashview_subcommands_live_upload_hashfile_job_pwdump(tmp_path):
) )
assert run.returncode == 0, output assert run.returncode == 0, output
assert ("Hashfile uploaded" in output) or ("Hashfile added" in output) assert ("Hashfile uploaded" in output) or ("Hashfile added" in output)
assert ("Job created" in output) or ("Failed to add job" in output) # Success surfaces the server's job id ("Job ID: N"); a graceful failure
# surfaces an "Error:" line. The CLI echoes the server's own message ("Job
# added"), so don't assert on the literal "Job created".
assert ("Job ID:" in output) or ("Error:" in output)
if "Job ID:" in output: if "Job ID:" in output:
job_id = None job_id = None
for line in output.splitlines(): for line in output.splitlines():
@@ -389,7 +405,10 @@ def test_hashview_subcommands_live_upload_hashfile_job_hashonly(tmp_path):
output = run.stdout + run.stderr output = run.stdout + run.stderr
assert run.returncode == 0, output assert run.returncode == 0, output
assert ("Hashfile uploaded" in output) or ("Hashfile added" in output) assert ("Hashfile uploaded" in output) or ("Hashfile added" in output)
assert ("Job created" in output) or ("Failed to add job" in output) # Success surfaces the server's job id ("Job ID: N"); a graceful failure
# surfaces an "Error:" line. The CLI echoes the server's own message ("Job
# added"), so don't assert on the literal "Job created".
assert ("Job ID:" in output) or ("Error:" in output)
if "Job ID:" in output: if "Job ID:" in output:
job_id = None job_id = None
for line in output.splitlines(): for line in output.splitlines():
+1 -1
View File
@@ -226,7 +226,7 @@ def test_get_customer_hashfiles_with_hashtype_filters(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
hv, hv,
"get_customer_hashfiles", "get_customer_hashfiles",
lambda customer_id: [ lambda customer_id, hash_type=None: [
{"customer_id": customer_id, "hashtype": "1000"}, {"customer_id": customer_id, "hashtype": "1000"},
{"customer_id": customer_id, "hash_type": "0"}, {"customer_id": customer_id, "hash_type": "0"},
], ],