mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-29 09:12:24 +01:00
245 lines
8.6 KiB
Python
245 lines
8.6 KiB
Python
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import zipfile
|
|
import pytest
|
|
|
|
from scripts.sanitize_trace_artifacts import (
|
|
sanitize_structured_obj,
|
|
sanitize_jsonl_content,
|
|
sanitize_url,
|
|
sanitize_body_text,
|
|
sanitize_trace_archive,
|
|
sanitize_directory,
|
|
)
|
|
|
|
|
|
def test_sanitize_headers_and_inverted_keys():
|
|
"""Inverted key order and non-Bearer schemes are safely redacted."""
|
|
# Standard order
|
|
h1 = {"name": "Authorization", "value": "Bearer my_super_secret_jwt_12345"}
|
|
assert sanitize_structured_obj(h1)["value"] == "[REDACTED]"
|
|
|
|
# Inverted order {"value": ..., "name": ...}
|
|
h2 = {"value": "Digest username=admin, nonce=abc, response=secret123", "name": "Authorization"}
|
|
res2 = sanitize_structured_obj(h2)
|
|
assert res2["value"] == "[REDACTED]"
|
|
|
|
# ApiKey scheme
|
|
h3 = {"name": "Authorization", "value": "ApiKey secret_api_key_xyz"}
|
|
assert sanitize_structured_obj(h3)["value"] == "[REDACTED]"
|
|
|
|
# Custom token header
|
|
h4 = {"name": "X-Auth-Token", "value": "auth_token_999"}
|
|
assert sanitize_structured_obj(h4)["value"] == "[REDACTED]"
|
|
|
|
|
|
def test_sanitize_cookies_arbitrary_names():
|
|
"""Arbitrary cookie names and values are completely redacted."""
|
|
cookie_entry = {
|
|
"cookies": [
|
|
{"name": "odysseus_session", "value": "secret_sess_abc"},
|
|
{"name": "remember_me_custom", "value": "remember_secret_val"},
|
|
{"name": "sso_auth_id", "value": "sso_uuid_12345"},
|
|
]
|
|
}
|
|
sanitized = sanitize_structured_obj(cookie_entry)
|
|
for c in sanitized["cookies"]:
|
|
assert c["value"] == "[REDACTED]"
|
|
|
|
|
|
def test_sanitize_dom_snapshots_and_nested_action_params():
|
|
"""DOM snapshot trees and deeply nested action params are redacted."""
|
|
# 1. Action with nested options before value
|
|
action_record = {
|
|
"type": "action",
|
|
"name": "fill",
|
|
"params": {
|
|
"selector": "#login-password",
|
|
"options": {"force": True, "timeout": 5000},
|
|
"value": "escaped\"quote\"password123!",
|
|
},
|
|
}
|
|
res_action = sanitize_structured_obj(action_record)
|
|
assert res_action["params"]["value"] == "[REDACTED]"
|
|
|
|
# 2. DOM snapshot input element
|
|
dom_record = {
|
|
"nodeName": "INPUT",
|
|
"attributes": {
|
|
"type": "password",
|
|
"name": "user_password",
|
|
"value": "super_secret_in_dom",
|
|
},
|
|
}
|
|
res_dom = sanitize_structured_obj(dom_record)
|
|
assert res_dom["attributes"]["value"] == "[REDACTED]"
|
|
|
|
|
|
def test_sanitize_url_tokens_and_authority():
|
|
"""URL query parameters and userinfo credentials are scrubbed."""
|
|
# Authority credentials
|
|
url1 = "https://admin:my_secret_pw@example.com/api/v1/resource"
|
|
assert sanitize_url(url1) == "https://admin:[REDACTED]@example.com/api/v1/resource"
|
|
|
|
# Query tokens
|
|
url2 = "https://example.com/oauth/callback?code=auth_code_secret&state=xyz&token=access_token_123"
|
|
sanitized2 = sanitize_url(url2)
|
|
assert "auth_code_secret" not in sanitized2
|
|
assert "access_token_123" not in sanitized2
|
|
assert "code=%5BREDACTED%5D" in sanitized2 or "code=[REDACTED]" in sanitized2
|
|
assert "token=%5BREDACTED%5D" in sanitized2 or "token=[REDACTED]" in sanitized2
|
|
assert "state=xyz" in sanitized2
|
|
|
|
|
|
def test_sanitize_request_bodies():
|
|
"""JSON and urlencoded form request bodies are scrubbed."""
|
|
# JSON body
|
|
json_body = '{"username": "admin", "password": "superSecretPassword123!", "api_key": "sk-123456789"}'
|
|
s_json = sanitize_body_text(json_body)
|
|
assert "superSecretPassword123!" not in s_json
|
|
assert "sk-123456789" not in s_json
|
|
assert '"password": "[REDACTED]"' in s_json
|
|
assert '"api_key": "[REDACTED]"' in s_json
|
|
assert '"username": "admin"' in s_json
|
|
|
|
# Form-encoded body
|
|
form_body = "username=alice&password=mypassword999&token=abcde&action=login"
|
|
s_form = sanitize_body_text(form_body)
|
|
assert "mypassword999" not in s_form
|
|
assert "abcde" not in s_form
|
|
assert "password=%5BREDACTED%5D" in s_form or "password=[REDACTED]" in s_form
|
|
assert "token=%5BREDACTED%5D" in s_form or "token=[REDACTED]" in s_form
|
|
assert "username=alice" in s_form
|
|
|
|
|
|
def test_sanitize_valid_archive(tmp_path):
|
|
"""Structured archive sanitization cleans HAR, trace, and masks source code without corruption."""
|
|
zp = str(tmp_path / "valid_trace.zip")
|
|
with zipfile.ZipFile(zp, "w") as z:
|
|
# JSON trace record
|
|
trace_record = {
|
|
"type": "action",
|
|
"name": "fill",
|
|
"params": {"selector": "#password", "value": 'secret"quote"pass'},
|
|
}
|
|
z.writestr("0-trace.trace", json.dumps(trace_record).encode() + b"\n")
|
|
|
|
# Network record with inverted header keys and URL query
|
|
net_record = {
|
|
"type": "request",
|
|
"request": {
|
|
"url": "https://localhost:7013/api/auth/login?token=token_abc",
|
|
"headers": [{"value": "Digest secret_digest", "name": "Authorization"}],
|
|
"postData": '{"password": "my_request_pw"}',
|
|
},
|
|
}
|
|
z.writestr("0-trace.network", json.dumps(net_record).encode() + b"\n")
|
|
|
|
# Embedded test source file
|
|
z.writestr("resources/src@test_auth.js", b'const password = "admin_password";\n')
|
|
|
|
# Binary image screenshot
|
|
z.writestr("screenshot.png", b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDRdiagnostic")
|
|
|
|
ok = sanitize_trace_archive(zp)
|
|
assert ok is True
|
|
assert os.path.exists(zp)
|
|
|
|
with zipfile.ZipFile(zp, "r") as z:
|
|
# Verify trace record
|
|
trace_data = json.loads(z.read("0-trace.trace").decode().strip())
|
|
assert trace_data["params"]["value"] == "[REDACTED]"
|
|
|
|
# Verify network record
|
|
net_data = json.loads(z.read("0-trace.network").decode().strip())
|
|
assert "token_abc" not in net_data["request"]["url"]
|
|
assert net_data["request"]["headers"][0]["value"] == "[REDACTED]"
|
|
assert "my_request_pw" not in net_data["request"]["postData"]
|
|
|
|
# Verify embedded source code was masked safely
|
|
src_content = z.read("resources/src@test_auth.js").decode()
|
|
assert "admin_password" not in src_content
|
|
assert "[SOURCE CODE REDACTED FOR TRACE CONFIDENTIALITY]" in src_content
|
|
assert "const password: [REDACTED]" not in src_content # No syntax error!
|
|
|
|
# Verify screenshot is unmodified
|
|
png = z.read("screenshot.png")
|
|
assert png == b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDRdiagnostic"
|
|
|
|
|
|
def test_fail_closed_malformed_zip(tmp_path):
|
|
zp = str(tmp_path / "corrupt.zip")
|
|
with open(zp, "wb") as f:
|
|
f.write(b"this is completely invalid zip data PK\x03\x04 corrupt")
|
|
|
|
ok = sanitize_trace_archive(zp)
|
|
assert ok is False
|
|
# Fail-closed: corrupted zip must be deleted immediately!
|
|
assert not os.path.exists(zp)
|
|
|
|
|
|
def test_fail_closed_partial_archive(tmp_path):
|
|
zp = str(tmp_path / "partial.zip")
|
|
with open(zp, "wb") as f:
|
|
f.write(b"PK\x03\x04\x14\x00\x00\x00")
|
|
|
|
ok = sanitize_trace_archive(zp)
|
|
assert ok is False
|
|
assert not os.path.exists(zp)
|
|
|
|
|
|
def test_fail_closed_empty_archive(tmp_path):
|
|
zp = str(tmp_path / "empty.zip")
|
|
with open(zp, "wb") as f:
|
|
pass
|
|
|
|
ok = sanitize_trace_archive(zp)
|
|
assert ok is False
|
|
assert not os.path.exists(zp)
|
|
|
|
|
|
def test_fail_closed_on_unhandled_exception(tmp_path, monkeypatch):
|
|
zp = str(tmp_path / "valid.zip")
|
|
with zipfile.ZipFile(zp, "w") as z:
|
|
z.writestr("test.trace", b'{"name": "fill"}\n')
|
|
|
|
def broken_sanitize(obj):
|
|
raise RuntimeError("Simulated crash")
|
|
|
|
monkeypatch.setattr("scripts.sanitize_trace_artifacts.sanitize_structured_obj", broken_sanitize)
|
|
|
|
ok = sanitize_trace_archive(zp)
|
|
assert ok is False
|
|
assert not os.path.exists(zp)
|
|
assert not os.path.exists(zp + ".tmp")
|
|
|
|
|
|
def test_multiple_trace_archives_in_directory(tmp_path):
|
|
sub1 = tmp_path / "test1"
|
|
sub2 = tmp_path / "test2"
|
|
sub1.mkdir()
|
|
sub2.mkdir()
|
|
|
|
z1 = str(sub1 / "trace.zip")
|
|
z2 = str(sub2 / "trace.zip")
|
|
|
|
with zipfile.ZipFile(z1, "w") as z:
|
|
z.writestr("0.trace", b'{"type": "action", "params": {"value": "secret1"}}\n')
|
|
with zipfile.ZipFile(z2, "w") as z:
|
|
z.writestr("0.trace", b'{"type": "action", "params": {"value": "secret2"}}\n')
|
|
|
|
count = sanitize_directory(str(tmp_path))
|
|
assert count == 2
|
|
|
|
assert os.path.exists(z1)
|
|
with zipfile.ZipFile(z1, "r") as z:
|
|
content = json.loads(z.read("0.trace").decode().strip())
|
|
assert content["params"]["value"] == "[REDACTED]"
|
|
|
|
assert os.path.exists(z2)
|
|
with zipfile.ZipFile(z2, "r") as z:
|
|
content = json.loads(z.read("0.trace").decode().strip())
|
|
assert content["params"]["value"] == "[REDACTED]"
|