318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""Credential-loss-prevention contracts for Atlas draft creation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from testing.tests.test_hermes_gitea_support import (
|
|
HEAD_SHA,
|
|
Response,
|
|
_draft_response,
|
|
_load,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sensitive",
|
|
[
|
|
"client_" + "secret=not-a-real-value",
|
|
'{"client_' + 'secret":\n"synthetic-value"}',
|
|
'{"client_' + 'secret"\n:\n"synthetic-value"}',
|
|
'{"client\\u005f' + 'secret":"synthetic-value"}',
|
|
"client_" + "secret:\n synthetic-value",
|
|
'{"client\\q' + 'secret":"synthetic-value"}',
|
|
'{"client\n' + 'secret":"synthetic-value"}',
|
|
'"client_' + 'secret"\x0b:\n"synthetic-value"',
|
|
"ACCESS_" + "TOKEN = 'not-a-real-value'",
|
|
'{"refresh_' + 'token": "not-a-real-value"}',
|
|
"private_" + "key: not-a-real-value",
|
|
"AWS_SECRET_ACCESS_" + "KEY=not-a-real-value",
|
|
"aws_access_key_" + "id: not-a-real-value",
|
|
"AWS_SESSION_" + "TOKEN = 'not-a-real-value'",
|
|
'{"AccessKey' + 'Id":"not-a-real-value"}',
|
|
'{"SecretAccess' + 'Key":"not-a-real-value"}',
|
|
'{"Session' + 'Token":"not-a-real-value"}',
|
|
"aws-security-" + "token: not-a-real-value",
|
|
"Account" + "Key=not-a-real-value",
|
|
"SharedAccess" + "Signature: not-a-real-value",
|
|
"AZURE_STORAGE_CONNECTION_" + "STRING='not-a-real-value'",
|
|
"DefaultEndpointsProtocol=https;AccountName=fake;Account"
|
|
+ "Key=not-a-real-value;EndpointSuffix=example",
|
|
"?sv=2024-11-04&ss=b&srt=sco&sp=rwdlac&se=2099-01-01&sig=" + "not-a-real-value",
|
|
"DOCKER_AUTH_" + "CONFIG='not-a-real-value'",
|
|
'{"auths":{"registry.example":{"auth":"bm90LXJlYWw="}}}',
|
|
'{"identity' + 'token":"not-a-real-value"}',
|
|
'{"type":"service_' + 'account","client_email":"fake@example.test"}',
|
|
'{"private_key_' + 'id":"not-a-real-value"}',
|
|
'{"client_' + 'email":"fake@example.test"}',
|
|
"GOOGLE_CREDENTIALS" + "=not-a-real-value",
|
|
"personal_access_" + "token: not-a-real-value",
|
|
"GITEA_" + "TOKEN=not-a-real-value",
|
|
"gitlab-token" + ": not-a-real-value",
|
|
"pat" + "=not-a-real-value",
|
|
"Authorization: " + "Bearer not-a-real-credential-value",
|
|
"authorization = " + '"Basic not-a-real-credential-value"',
|
|
"Bearer" + "=not-a-real-credential-value",
|
|
"Basic" + ": not-a-real-credential-value",
|
|
"pass" + "word=not-a-real-credential",
|
|
"ghp_" + "notarealcredentialvalue123456",
|
|
"github_pat_" + "notarealcredentialvalue123456",
|
|
"glpat-" + "notarealcredentialvalue123456",
|
|
"xoxb-" + "not-a-real-credential-value-123456",
|
|
"sk-ant-" + "notarealcredentialvalue123456",
|
|
"sk-proj-" + "notarealcredentialvalue123456",
|
|
"sk_live_" + "notarealcredentialvalue123456",
|
|
"ya29." + "notarealcredentialvalue123456",
|
|
"gta_" + "notarealcredentialvalue123456",
|
|
"whsec_" + "notarealcredentialvalue123456",
|
|
"npm_" + "notarealcredentialvalue123456",
|
|
"pypi-" + "notarealcredentialvalue123456789012345",
|
|
"hf_" + "notarealcredentialvalue123456",
|
|
"SG." + "notarealvalue1234" + ".notarealcredentialvalue123456",
|
|
"SK" + "a" * 32,
|
|
"https://hooks.slack.com/services/" + "T000/B000/notarealvalue123456",
|
|
"https://discord.com/api/webhooks/123456789/" + "notarealcredentialvalue123456",
|
|
"https://fake.webhook.office.com/" + "notarealcredentialvalue123456",
|
|
"webhook_" + "url=https://example.test/not-real",
|
|
"FutureCloudSigning" + "Credential=not-a-real-value",
|
|
"future-client-signing-" + "key: not-a-real-value",
|
|
"future_client_signing_" + "key='not-a-real-value'",
|
|
'{"serviceAccountPrivate' + 'Key":"not-a-real-value"}',
|
|
"CONTAINER_REGISTRY_" + "CREDENTIAL=not-a-real-value",
|
|
"someWebhookSigning" + "Secret: not-a-real-value",
|
|
"client" + "Key=not-a-real-value",
|
|
"session" + "Key: not-a-real-value",
|
|
"access" + "Id=not-a-real-value",
|
|
"credentials" + ": {user: fake}",
|
|
"private" + "Key: |",
|
|
"nuget_api_" + "key=not-a-real-value",
|
|
"oy2" + "a" * 44,
|
|
"sk_test_" + "notarealcredentialvalue123456",
|
|
"A1b2C3d4E5f6G7h8I9j0K_l-M+n/O=pQ2rS3tU4vW5xY6zZ7aB8cC9d",
|
|
"AKIA" + "A" * 16,
|
|
"AIza" + "a" * 35,
|
|
"-----BEGIN OPENSSH " + "PRIVATE KEY-----",
|
|
"ssh-ed25519 " + "bm90YXJlYWxjcmVkZW50aWFsdmFsdWU=",
|
|
"eyJnotarealheader." + "notarealpayloadvalue." + "notarealsignature",
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("field", ["title", "body"])
|
|
def test_create_rejects_expanded_credential_shapes_before_network(
|
|
sensitive: str, field: str
|
|
):
|
|
client = _load()
|
|
called = False
|
|
request_built = False
|
|
|
|
original_build_request = client.build_request
|
|
|
|
def build_request(*args, **kwargs):
|
|
nonlocal request_built
|
|
request_built = True
|
|
return original_build_request(*args, **kwargs)
|
|
|
|
def opener(*_args, **_kwargs):
|
|
nonlocal called
|
|
called = True
|
|
return Response(_draft_response())
|
|
|
|
client.build_request = build_request
|
|
values = {"title": "Focused fix", "body": "Review evidence"}
|
|
values[field] = sensitive
|
|
|
|
with pytest.raises(client.PolicyError, match="credential material"):
|
|
client.create_draft(
|
|
"cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title=values["title"],
|
|
body=values["body"],
|
|
token="runtime-sentinel",
|
|
opener=opener,
|
|
)
|
|
assert request_built is False
|
|
assert called is False
|
|
|
|
|
|
def test_very_long_compact_body_is_rejected_before_request_or_network():
|
|
client = _load()
|
|
request_built = False
|
|
opener_called = False
|
|
original_build_request = client.build_request
|
|
|
|
def build_request(*args, **kwargs):
|
|
nonlocal request_built
|
|
request_built = True
|
|
return original_build_request(*args, **kwargs)
|
|
|
|
def opener(*_args, **_kwargs):
|
|
nonlocal opener_called
|
|
opener_called = True
|
|
return Response(_draft_response())
|
|
|
|
client.build_request = build_request
|
|
with pytest.raises(client.PolicyError, match="credential material"):
|
|
client.create_draft(
|
|
"cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title="Focused fix",
|
|
body="a" * 300,
|
|
token="runtime-sentinel",
|
|
opener=opener,
|
|
)
|
|
assert request_built is False
|
|
assert opener_called is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"safe_text",
|
|
[
|
|
"AWS_SECRET_ACCESS_KEY is injected at runtime",
|
|
"Token: reject empty values",
|
|
"Authorization = preserve header behavior",
|
|
"Password: add regression",
|
|
"Review the Authorization header behavior",
|
|
"Bearer authentication is required for this route",
|
|
"Document Docker auths payload rejection",
|
|
"The client_email field belongs to service accounts",
|
|
"AccountKey assignments must be blocked",
|
|
"This patch changes token validation without including a value",
|
|
"FutureCloudSigningCredential handling needs a regression test",
|
|
"The clientKey name is documented without an assigned value",
|
|
"A SHA-256 digest 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef is evidence",
|
|
],
|
|
)
|
|
def test_credential_policy_keeps_normal_engineering_prose_usable(safe_text: str):
|
|
client = _load()
|
|
|
|
assert client._validate_body(safe_text) == safe_text
|
|
assert client._draft_title(safe_text) == f"WIP: {safe_text}"
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["title", "body"])
|
|
def test_create_rejects_exact_runtime_token_before_network(field: str):
|
|
client = _load()
|
|
called = False
|
|
request_built = False
|
|
|
|
original_build_request = client.build_request
|
|
|
|
def build_request(*args, **kwargs):
|
|
nonlocal request_built
|
|
request_built = True
|
|
return original_build_request(*args, **kwargs)
|
|
|
|
def opener(*_args, **_kwargs):
|
|
nonlocal called
|
|
called = True
|
|
return Response(_draft_response())
|
|
|
|
client.build_request = build_request
|
|
runtime_token = "exact-random-runtime-sentinel-7b73ac61"
|
|
values = {"title": "Focused fix", "body": "Review evidence"}
|
|
values[field] = f"Accidental {runtime_token} value"
|
|
with pytest.raises(client.PolicyError, match="runtime credential"):
|
|
client.create_draft(
|
|
"cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title=values["title"],
|
|
body=values["body"],
|
|
token=runtime_token,
|
|
opener=opener,
|
|
)
|
|
assert request_built is False
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["repo", "base", "head", "title", "body"])
|
|
def test_every_public_field_rejects_exact_runtime_token_before_git_or_network(
|
|
field: str, monkeypatch
|
|
):
|
|
client = _load()
|
|
git_called = False
|
|
request_built = False
|
|
opener_called = False
|
|
runtime_token = "runtime-sentinel"
|
|
values = {
|
|
"repo": "cassandra",
|
|
"base": "main",
|
|
"head": "hermes/fix",
|
|
"title": "Focused fix",
|
|
"body": "Review evidence",
|
|
}
|
|
values[field] = runtime_token
|
|
|
|
def git_run(*_args, **_kwargs):
|
|
nonlocal git_called
|
|
git_called = True
|
|
raise AssertionError("Git must not run for a credential-bearing field")
|
|
|
|
def build_request(*_args, **_kwargs):
|
|
nonlocal request_built
|
|
request_built = True
|
|
raise AssertionError("a request must not be built")
|
|
|
|
def opener(*_args, **_kwargs):
|
|
nonlocal opener_called
|
|
opener_called = True
|
|
raise AssertionError("the opener must not be called")
|
|
|
|
monkeypatch.setattr(client._validate_ref.__globals__["subprocess"], "run", git_run)
|
|
monkeypatch.setattr(client, "build_request", build_request)
|
|
with pytest.raises(client.PolicyError, match="runtime credential"):
|
|
client.create_draft(
|
|
values["repo"],
|
|
base=values["base"],
|
|
head=values["head"],
|
|
head_sha=HEAD_SHA,
|
|
title=values["title"],
|
|
body=values["body"],
|
|
token=runtime_token,
|
|
opener=opener,
|
|
)
|
|
assert git_called is False
|
|
assert request_built is False
|
|
assert opener_called is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sensitive",
|
|
[
|
|
" ".join(["{}"] * 32) + ' {"client_secret":"synthetic-value"}',
|
|
"prefix_" + "ghp_" + "notarealcredentialvalue123456_suffix",
|
|
"client_secret: correct horse battery staple",
|
|
"'client_secret':\n synthetic-value",
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("field", ["title", "body"])
|
|
def test_structured_scanner_closes_bounded_and_wrapped_token_bypasses(
|
|
sensitive: str, field: str
|
|
):
|
|
client = _load()
|
|
values = {"title": "Focused fix", "body": "Review evidence"}
|
|
values[field] = sensitive
|
|
called = False
|
|
|
|
def opener(*_args, **_kwargs):
|
|
nonlocal called
|
|
called = True
|
|
return Response(_draft_response())
|
|
|
|
with pytest.raises(client.PolicyError, match="credential material"):
|
|
client.create_draft(
|
|
"cassandra",
|
|
base="main",
|
|
head="hermes/fix",
|
|
head_sha=HEAD_SHA,
|
|
title=values["title"],
|
|
body=values["body"],
|
|
token="runtime-sentinel",
|
|
opener=opener,
|
|
)
|
|
assert called is False
|