Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions src/kiro_crew/testing/fake_computer_use.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,12 +88,10 @@
FAKE_IMAGE_HEIGHT = 1

# ── Redaction fixtures ──
# The canonical PUBLIC documentation access-key id (it appears verbatim in AWS's
# own docs and in this repo's redaction tests) plus two exfiltration-shaped URLs
# using IANA-reserved documentation ranges. None of these is a real credential or
# a reachable host; they exist so a test can assert that the rendered tree comes
# back MASKED.
FAKE_CREDENTIAL_FIXTURE = "AKIA" "IOSFODNN7EXAMPLE"
# A GitLab-shaped fake token plus two exfiltration-shaped URLs using IANA-reserved
# documentation ranges. None of these is a real credential or a reachable host;
# they exist so a test can assert that the rendered tree comes back MASKED.
FAKE_CREDENTIAL_FIXTURE = "glpat-" "IOSFODNN7EXAMPLE"
FAKE_EXFIL_URL = (
"https://evil.example.com/collect?data="
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
Expand Down
12 changes: 6 additions & 6 deletions test/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -390,19 +390,19 @@ def forget_env_at_teardown(monkeypatch, *names: str) -> None:
#: a cut may fall (``test_display_split_safety.py``) and the renderer that applies
#: it (``test_wecom_renderer.py``) -- and a duplicated fixture table drifts.
CREDENTIAL_STRADDLE_SHAPES = [
pytest.param("[AKIA](https://ex.test/a,b)", "IOSFODNN7EXAMPLE", id="link-target-comma"),
pytest.param("[glpat-](https://ex.test/a,b)", "IOSFODNN7EXAMPLE", id="link-target-comma"),
pytest.param(
"Authorization: Bearer",
" abcdefghijklmnopqrstuvwxyz0123456789",
id="header-whitespace",
),
pytest.param("AKIAIOSF_", "_ODNN7EXAMPLE", id="underscore-emphasis"),
pytest.param("AKIAIOSF~", "~ODNN7EXAMPLE", id="tilde-emphasis"),
pytest.param("https://evil.test/?q=AKIAIOSFODNN7", "EXAMPLE.", id="url-punctuation"),
pytest.param("[l](https://ex.test/x/AKIAIOSF", "ODNN7EXAMPLE)", id="cut-inside-a-url"),
pytest.param("glpat-IOSF_", "_ODNN7EXAMPLE", id="underscore-emphasis"),
pytest.param("glpat-IOSF~", "~ODNN7EXAMPLE", id="tilde-emphasis"),
pytest.param("https://evil.test/?q=glpat-IOSFODNN7", "EXAMPLE.", id="url-punctuation"),
pytest.param("[l](https://ex.test/x/glpat-IOSF", "ODNN7EXAMPLE)", id="cut-inside-a-url"),
pytest.param(f"{_PEM_DASHES}BEGIN RSA PRIV", _PEM_TAIL_HALF, id="pem-anchor"),
pytest.param(f"{_PEM_DASHES}BEG**IN** RSA PRIV", _PEM_TAIL_HALF, id="pem-anchor-markup-split"),
pytest.param("AKIAIOSF", "ODNN7EXAMPLE", id="no-markup-at-all"),
pytest.param("glpat-IOSF", "ODNN7EXAMPLE", id="no-markup-at-all"),
]


Expand Down
2 changes: 1 addition & 1 deletion test/discord_resumed_busy_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
# The refusal every case in the test_discord_resumed_busy*.py suites either avoids or keeps.
_REFUSAL = "busy with a turn started elsewhere"
# A credential the display redaction rewrites, inside otherwise ordinary text.
_SECRET = "AKIAIOSFODNN7EXAMPLE"
_SECRET = "glpat-IOSFODNN7EXAMPLE"
_TYPED = f"use https://docs.example.com/page?key={_SECRET} for this"


Expand Down
8 changes: 4 additions & 4 deletions test/markdown_link_parentheses_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ def _regex_destination(rest: str) -> tuple[str | None, int]:

#: The AWS example access key, split by a link so that only a reader who sees the
#: link rendered away sees it whole.
_KEY = "AKIAIOSFODNN7EXAMPLE"
_KEY = "glpat-IOSFODNN7EXAMPLE"


def _screen_readings(text: str) -> list[str]:
Expand All @@ -79,12 +79,12 @@ def _discord_reader_sees(text: str) -> str:

#: The trigger that makes ``redact_for_display`` emit the canonical form: a link
#: whose url holds the key split by ``**``, whole on a plain-text sink only.
_URL_SPLIT_TRIGGER = f"[l](https://x/AKIA**{_KEY[4:]}**)"
_URL_SPLIT_TRIGGER = f"[l](https://x/glpat-**{_KEY[6:]}**)"


def _nested_link(depth: int) -> str:
"""``[AKIA](u)`` wrapped in *depth* more ``[...](u)`` links, one level per pass."""
link = "[AKIA](u)"
"""``[glpat-](u)`` wrapped in *depth* more ``[...](u)`` links, one level per pass."""
link = "[glpat-](u)"
for _ in range(depth):
link = f"[{link}](u)"
return link
2 changes: 1 addition & 1 deletion test/member_memory_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
from kiro_crew.vector_memory import VectorMemoryStore, create_member_database, open_member_database

#: A credential-shaped token for "sensitive document" fixtures.
DOCUMENT_CREDENTIAL = "AKIAIOSFODNN7EXAMPLE"
DOCUMENT_CREDENTIAL = "glpat-IOSFODNN7EXAMPLE"

MEMBERS = ("alice", "bob")

Expand Down
2 changes: 1 addition & 1 deletion test/metrics/test_recorder.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ def test_credential_attr_redacted(self):
rec.counter(
"kirocrew.test.count",
1,
attrs={"api_key": "AKIAIOSFODNN7EXAMPLE1", "safe": "ok"},
attrs={"api_key": "glpat-IOSFODNN7EXAMPLE1", "safe": "ok"},
)
points = _data_points(reader)
assert points, "metric should have recorded"
Expand Down
8 changes: 4 additions & 4 deletions test/metrics/test_telemetry_titles.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,11 +69,11 @@ def _cost(*slot_keys: str) -> dict:
async def test_a_credential_shaped_title_is_redacted():
# A real AWS-key shape. The scanner recognises it; the point is that this
# endpoint runs the scanner at all.
leaked = "notes for AKIA" + "IOSFODNN7EXAMPLE"
leaked = "notes for glpat-" + "IOSFODNN7EXAMPLE"
slots = {"chat-1-1": SimpleNamespace(display_title=leaked)}
out = await _with_conversation_titles(_request(slots), _cost("chat-1-1"))
title = out["conversations"][0]["title"]
assert "AKIA" + "IOSFODNN7EXAMPLE" not in title
assert "glpat-" + "IOSFODNN7EXAMPLE" not in title
assert "REDACTED" in title


Expand Down Expand Up @@ -188,15 +188,15 @@ async def test_a_transcript_without_an_explicit_title_yields_none(self, tmp_path

@pytest.mark.asyncio
async def test_a_persisted_title_is_redacted_like_a_live_one(self, tmp_path):
leaked = "notes for AKIA" + "IOSFODNN7EXAMPLE"
leaked = "notes for glpat-" + "IOSFODNN7EXAMPLE"
log = ConversationLog(tmp_path)
log.set_title(slot_transcript_key("chat-1-1"), leaked)

out = await _with_conversation_titles(
_request({}, conversation_log=log), _cost("chat-1-1")
)
title = out["conversations"][0]["title"]
assert "AKIA" + "IOSFODNN7EXAMPLE" not in title
assert "glpat-" + "IOSFODNN7EXAMPLE" not in title
assert "REDACTED" in title

@pytest.mark.asyncio
Expand Down
2 changes: 1 addition & 1 deletion test/split_credential_boundary_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

#: AWS's own documented example key, the same fixture the display-safety tests
#: use, so no real credential shape is introduced anywhere.
KEY = "AKIAIOSFODNN7EXAMPLE"
KEY = "glpat-IOSFODNN7EXAMPLE"
HEAD, TAIL = KEY[:10], KEY[10:]

SRC = Path(kiro_crew.__file__).parent
Expand Down
2 changes: 1 addition & 1 deletion test/telegram_parity_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
from kiro_crew.telegram.transport import TELEGRAM_CAPABILITIES, TelegramInboundMessage

# Split so the literal never appears whole in this file.
_AWS_KEY = "AKIA" + "IOSFODNN7EXAMPLE"
_AWS_KEY = "glpat-" + "IOSFODNN7EXAMPLE"


@pytest.fixture(autouse=True)
Expand Down
22 changes: 11 additions & 11 deletions test/test_acp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -540,20 +540,20 @@ def test_credentials_in_data_are_redacted(self):
err = {
"code": -32603,
"message": "Internal error",
"data": "AccessDenied: AKIAIOSFODNN7EXAMPLE not authorized",
"data": "AccessDenied: glpat-IOSFODNN7EXAMPLE not authorized",
}
out = _format_acp_error(err)
assert "AKIAIOSFODNN7EXAMPLE" not in out
assert "glpat-IOSFODNN7EXAMPLE" not in out

def test_credentials_in_unknown_dict_fallback_are_redacted(self):
"""The fallback path echoes raw dict — must still scrub secrets."""
err = {
"code": -32000,
"message": "weird upstream",
"data": "leak: AKIAIOSFODNN7EXAMPLE",
"data": "leak: glpat-IOSFODNN7EXAMPLE",
}
out = _format_acp_error(err)
assert "AKIAIOSFODNN7EXAMPLE" not in out
assert "glpat-IOSFODNN7EXAMPLE" not in out
# The scrubbed placeholder proves redaction ran on this path.
assert "REDACTED" in out
# The surrounding provider text still reaches the user.
Expand Down Expand Up @@ -591,14 +591,14 @@ def test_sensitive_content_emits_log_warning(self, caplog):
err = {
"code": -32000,
"message": "weird upstream",
"data": "leak: AKIAIOSFODNN7EXAMPLE",
"data": "leak: glpat-IOSFODNN7EXAMPLE",
}
with caplog.at_level(logging.WARNING, logger="kiro_crew.acp.client"):
_format_acp_error(err)

warnings = [r for r in caplog.records if "sensitive content" in r.getMessage()]
assert warnings, "expected a redaction warning to be logged"
assert "AKIAIOSFODNN7EXAMPLE" not in warnings[0].getMessage()
assert "glpat-IOSFODNN7EXAMPLE" not in warnings[0].getMessage()

def test_internal_server_error_rewrite(self):
"""A transient 5xx from the backend must
Expand Down Expand Up @@ -667,10 +667,10 @@ def test_transient_branch_does_not_leak_credentials(self):
err = {
"code": -32603,
"message": "Internal error",
"data": "InternalServerError: leaked AKIAIOSFODNN7EXAMPLE please try again",
"data": "InternalServerError: leaked glpat-IOSFODNN7EXAMPLE please try again",
}
out = _format_acp_error(err)
assert "AKIAIOSFODNN7EXAMPLE" not in out
assert "glpat-IOSFODNN7EXAMPLE" not in out
assert "transient error" in out.lower()

def test_throttle_precedence_over_transient(self):
Expand Down Expand Up @@ -990,9 +990,9 @@ def test_is_bounded_and_redacted(self):
assert len(detail) == _COMPACTION_DETAIL_MAX_CHARS

secret = compaction_failure_detail(
{"status": {"type": "failed", "error": "aws_secret_access_key=AKIAIOSFODNN7EXAMPLE"}}
{"status": {"type": "failed", "error": "aws_secret_access_key=glpat-IOSFODNN7EXAMPLE"}}
)
assert "AKIAIOSFODNN7EXAMPLE" not in secret
assert "glpat-IOSFODNN7EXAMPLE" not in secret

def test_rejects_a_placeholder_reason(self):
"""A named reason of "error" is not a reason. KAS's summarization_failed
Expand Down Expand Up @@ -1043,7 +1043,7 @@ def test_the_frame_is_redacted_before_it_reaches_the_log(self, caplog):
import logging

handle = self._bare_handle()
secret = "AKIAIOSFODNN7EXAMPLE"
secret = "glpat-IOSFODNN7EXAMPLE"
update = {
"sessionUpdate": "session_info_update",
"_meta": {
Expand Down
4 changes: 2 additions & 2 deletions test/test_acp_client_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -225,14 +225,14 @@ async def test_failed_with_a_credential_shaped_summary_is_redacted(self, tmp_pat
method=METHOD_COMPACTION_STATUS,
params={
"status": {"type": "failed"},
"summary": "backend error: key AKIAIOSFODNN7EXAMPLE rejected",
"summary": "backend error: key glpat-IOSFODNN7EXAMPLE rejected",
},
)
client._read_message = AsyncMock(return_value=msg)

result = await client.wait_for_compaction(timeout=5.0)
assert result["type"] == "failed"
assert "AKIAIOSFODNN7EXAMPLE" not in result["summary"]
assert "glpat-IOSFODNN7EXAMPLE" not in result["summary"]
assert "[REDACTED: credential]" in result["summary"]

@pytest.mark.asyncio
Expand Down
8 changes: 4 additions & 4 deletions test/test_acp_client_inbound.py
Original file line number Diff line number Diff line change
Expand Up @@ -1078,7 +1078,7 @@ class TestDrainStderrRedaction:
@pytest.mark.asyncio
async def test_raw_stored_redacted_logged(self):
client = AcpClient()
raw = "Error: key=AKIAIOSFODNN7EXAMPLE connect failed"
raw = "Error: key=glpat-IOSFODNN7EXAMPLE connect failed"

reader = AsyncMock(spec=["readline"])
reader.readline = AsyncMock(side_effect=[raw.encode() + b"\n", b""])
Expand All @@ -1090,7 +1090,7 @@ async def test_raw_stored_redacted_logged(self):
assert list(client._stderr_lines) == [raw]
# Logged call used redacted text (credential replaced)
logged_text = mock_logger.warning.call_args[0][2]
assert "AKIAIOSFODNN7EXAMPLE" not in logged_text
assert "glpat-IOSFODNN7EXAMPLE" not in logged_text


class TestDrainStderrSuppression:
Expand Down Expand Up @@ -1177,14 +1177,14 @@ async def test_acperror_contains_redacted_stderr(self):
client._process.returncode = 1
client._process.stdout = MagicMock()
client._process.stdout.readline = AsyncMock(return_value=b"")
client._stderr_lines = deque(["secret key=AKIAIOSFODNN7EXAMPLE"])
client._stderr_lines = deque(["secret key=glpat-IOSFODNN7EXAMPLE"])
client._stderr_task = MagicMock()
client._stderr_task.done.return_value = True

with pytest.raises(AcpError, match="ACP process exited") as exc_info:
await client._read_message(timeout=1.0)

assert "AKIAIOSFODNN7EXAMPLE" not in str(exc_info.value)
assert "glpat-IOSFODNN7EXAMPLE" not in str(exc_info.value)


class TestTrackMetadataCredits:
Expand Down
20 changes: 10 additions & 10 deletions test/test_acp_client_model_substitution.py
Original file line number Diff line number Diff line change
Expand Up @@ -557,11 +557,11 @@ async def test_new_session_following_substitution_redacts_log_payload(self, tmp_

# Inject a substitute that embeds an AWS access-key-shaped substring
# so the dual redactor heuristic actually fires (redact_exfiltration_urls
# matches the URL because the query carries an AKIA token, and
# redact_credentials matches the standalone AKIA pattern). A bare
# matches the URL because the query carries a glpat- token, and
# redact_credentials matches the standalone glpat- pattern). A bare
# arbitrary URL is too tame for the heuristic and would leave the test
# asserting on a no-op pass-through.
url_shaped = "http://attacker.example.com/exfil?token=AKIAIOSFODNN7EXAMPLE"
url_shaped = "http://attacker.example.com/exfil?token=glpat-IOSFODNN7EXAMPLE"

sent: list[str] = []

Expand Down Expand Up @@ -598,7 +598,7 @@ async def _wait(req_id, timeout=0.0, *, method="", expected_mcp=None):
# inside "[REDACTED: suspicious URL to <host>]" so operators can
# investigate the leak source -- that is BY DESIGN, not a bypass.
assert (
"AKIAIOSFODNN7EXAMPLE" not in emitted
"glpat-IOSFODNN7EXAMPLE" not in emitted
), "AWS access-key substring MUST be redacted before logging"
assert (
"REDACTED" in emitted
Expand All @@ -620,7 +620,7 @@ async def test_initialize_session_redacts_model_in_info_log(self, tmp_path):
client._process.returncode = None
# AWS access-key-shaped substring guarantees both redactors fire (URL
# query matches _EXFIL_PATTERNS; standalone key matches _CREDENTIAL_PATTERNS).
client._model = "http://attacker.example.com/m?token=AKIAIOSFODNN7EXAMPLE"
client._model = "http://attacker.example.com/m?token=glpat-IOSFODNN7EXAMPLE"

async def _send(method, params):
return 1
Expand Down Expand Up @@ -658,7 +658,7 @@ async def _drain():
# See sibling test for the design note: redactor keeps the host as a
# diagnostic label inside "[REDACTED: suspicious URL to <host>]".
assert (
"AKIAIOSFODNN7EXAMPLE" not in emitted_model
"glpat-IOSFODNN7EXAMPLE" not in emitted_model
), "AWS access-key substring MUST be redacted before logging"
assert (
"REDACTED" in emitted_model
Expand Down Expand Up @@ -694,7 +694,7 @@ async def _no_session_with_credshaped(*args, **kwargs):
# Helper adopted a credential-shaped substitute then gave up without
# producing a sessionId. Drives the "even after adopting substitute
# model X" branch where X is credential-shaped.
client._model = "AKIAIOSFODNN7EXAMPLE"
client._model = "glpat-IOSFODNN7EXAMPLE"
return {}

client._new_session_following_substitution = _no_session_with_credshaped # type: ignore[assignment]
Expand All @@ -704,7 +704,7 @@ async def _no_session_with_credshaped(*args, **kwargs):

msg = str(excinfo.value)
assert (
"AKIAIOSFODNN7EXAMPLE" not in msg
"glpat-IOSFODNN7EXAMPLE" not in msg
), "AWS access-key-shaped substitute MUST be redacted before reaching the AcpError message"
assert "even after adopting substitute model" in msg

Expand All @@ -718,7 +718,7 @@ async def test_wait_for_response_catchall_raise_redacts_msg_error(self, tmp_path
real_err = {
"code": -32603,
"message": "Internal error",
"data": "request_id: AKIAIOSFODNN7EXAMPLE backend trace failed",
"data": "request_id: glpat-IOSFODNN7EXAMPLE backend trace failed",
}
msg = MagicMock()
msg.is_response_for.return_value = True
Expand All @@ -736,7 +736,7 @@ async def _one_message(timeout=0.0):
await client._wait_for_response(13, timeout=1.0)

emitted = str(excinfo.value)
assert "AKIAIOSFODNN7EXAMPLE" not in emitted, (
assert "glpat-IOSFODNN7EXAMPLE" not in emitted, (
"AWS access-key-shaped substring in msg.error MUST be redacted "
"before reaching the catch-all AcpError message"
)
Expand Down
10 changes: 5 additions & 5 deletions test/test_acp_client_prompt_turn.py
Original file line number Diff line number Diff line change
Expand Up @@ -917,7 +917,7 @@ async def test_dispatch_redacts_and_extracts_subagent_output(self):
"update": {
"sessionUpdate": "tool_call_chunk",
"toolCallId": "tc-1",
"title": "leak AKIAIOSFODNN7EXAMPLE now",
"title": "leak glpat-IOSFODNN7EXAMPLE now",
},
}
),
Expand All @@ -930,7 +930,7 @@ async def test_dispatch_redacts_and_extracts_subagent_output(self):
"sessionId": "s1",
"update": {
"sessionUpdate": "agent_message_chunk",
"content": {"text": "secret AKIAIOSFODNN7EXAMPLE here"},
"content": {"text": "secret glpat-IOSFODNN7EXAMPLE here"},
},
}
),
Expand All @@ -951,13 +951,13 @@ async def _fake_loop(req_id, timeout):
acts = [e for e in events if e.kind == EVENT_SUBAGENT_ACTIVITY]
tc = [a for a in acts if a.tool_call_id == "tc-1"]
txt = [a for a in acts if a.text]
# Title redacted (credential scrubbed, no raw AKIA leaks through).
# Title redacted (credential scrubbed, no raw glpat- leaks through).
assert len(tc) == 1
assert "AKIAIOSFODNN7EXAMPLE" not in tc[0].title
assert "glpat-IOSFODNN7EXAMPLE" not in tc[0].title
assert "[REDACTED" in tc[0].title
# Nested content.text was extracted (robustness) AND redacted (security).
assert len(txt) == 1 and txt[0].sub_session_id == "s1"
assert "AKIAIOSFODNN7EXAMPLE" not in txt[0].text
assert "glpat-IOSFODNN7EXAMPLE" not in txt[0].text
assert "[REDACTED" in txt[0].text

@pytest.mark.asyncio
Expand Down
Loading
Loading