import hashlib import json import re from dataclasses import dataclass, field MAX_CANDIDATE_SECRET_BYTES = 1024 * 1024 MAX_CANDIDATE_METADATA_BYTES = 64 * 1024 @dataclass(frozen=True) class CandidateSpec: service: str candidate_kind: str credential_hash: str provider_key_hash: str secret_hash: str secret_text: str | None = None secret_json: str | None = None key_masked: str = '' endpoint: str = '' principal: str = '' metadata: dict = field(default_factory=dict) def as_frame(self, attribution=None): return { 'service': self.service, 'candidate_kind': self.candidate_kind, 'credential_hash': self.credential_hash, 'provider_key_hash': self.provider_key_hash, 'secret_hash': self.secret_hash, 'secret_text': self.secret_text, 'secret_json': self.secret_json, 'key_masked': self.key_masked, 'endpoint': self.endpoint, 'principal': self.principal, 'metadata': self.metadata, 'attribution': dict(attribution or {}), } DETECTOR_SERVICES = { 'openai': 'openai', 'anthropic': 'anthropic', 'qwendashscope': 'qwen', 'qwen_dashscope': 'qwen', 'qwen': 'qwen', 'dashscope': 'qwen', 'deepseek': 'deepseek', 'deepseekapikey': 'deepseek', 'deepseek_api_key': 'deepseek', 'zaiglm': 'zai', 'kimimoonshot': 'kimi', 'moonshotai': 'kimi', 'moonshot': 'kimi', 'kimi': 'kimi', 'openrouter': 'openrouter', 'groq': 'groq', 'replicate': 'replicate', 'xai': 'xai', 'huggingface': 'huggingface', 'github': 'github', 'githuboauth2': 'github', 'gitlab': 'gitlab', 'aws': 'aws', 'gcp': 'gcp', 'gcpapplicationdefaultcredentials': 'gcp', 'googleai': 'gemini', 'googleaistudio': 'gemini', 'azure': 'azure', 'azureopenai': 'azure', 'azurecontainerregistry': 'azure', 'azurefoundryendpointbeforekey': 'azure', 'azurefoundrykeybeforeendpoint': 'azure', 'dockerhub': 'dockerhub', } GEMINI_KEY_RE = re.compile( r'AIza[0-9A-Za-z_-]{20,}|(? max_bytes: raise ValueError('keycheck candidate field exceeds its byte bound') return text def _mask(value): text = str(value or '') if len(text) <= 8: return '*' * len(text) return text[:4] + ('*' * min(24, len(text) - 8)) + text[-4:] def _service_for_finding(finding): context = finding.get('ScannerContext') if isinstance(finding.get('ScannerContext'), dict) else {} hint = str(context.get('provider_hint') or '').lower() if hint in AMBIGUOUS_PROVIDER_HINTS: return 'provider_resolver' if hint in GENERIC_SK_PROVIDERS: return hint detector = re.sub(r'[^a-z0-9_]', '', str( finding.get('DetectorName') or finding.get('DetectorType') or '' ).lower()) service = DETECTOR_SERVICES.get(detector, '') if service: return service extra = finding.get('ExtraData') if isinstance(finding.get('ExtraData'), dict) else {} name = re.sub(r'[^a-z0-9_]', '', str(extra.get('name') or '').lower()) return DETECTOR_SERVICES.get(name, '') def _make_candidate( service, candidate_kind, probe_material, raw_material, *, secret_text=None, secret_json=None, endpoint='', principal='', metadata=None, ): probe_material = _bounded(probe_material, MAX_CANDIDATE_SECRET_BYTES) raw_material = _bounded(raw_material, MAX_CANDIDATE_SECRET_BYTES) provider_key_hash = hashlib.sha256(probe_material.encode('utf-8')).hexdigest() secret_hash = hashlib.sha256(raw_material.encode('utf-8')).hexdigest() credential_hash = hashlib.sha256('|'.join(( 'truf-credential-v2', service, probe_material, )).encode('utf-8')).hexdigest() metadata = dict(metadata or {}) if len(_json(metadata).encode('utf-8')) > MAX_CANDIDATE_METADATA_BYTES: raise ValueError('keycheck candidate metadata exceeds its byte bound') return CandidateSpec( service=service, candidate_kind=candidate_kind, credential_hash=credential_hash, provider_key_hash=provider_key_hash, secret_hash=secret_hash, secret_text=secret_text, secret_json=secret_json, key_masked=_mask(probe_material), endpoint=endpoint, principal=principal, metadata=metadata, ) def stored_provider_key_hash(service, candidate_kind, secret_text, secret_json, endpoint='', principal=''): service = str(service or '').lower() candidate_kind = str(candidate_kind or '') secret_text = str(secret_text or '') secret_json = str(secret_json or '') endpoint = str(endpoint or '').lower() principal = str(principal or '') if service == 'gcp': parsed = _json_object(secret_json) probe = _provider_json(parsed) if parsed is not None else secret_json elif service == 'azure' and candidate_kind == 'azure_service_principal': parsed = _json_object(secret_json) or {} probe = ':'.join(str(parsed.get(name) or '') for name in ('tenantId', 'clientId', 'clientSecret')) if probe == '::': probe = ':'.join(str(parsed.get(name) or '') for name in ('tenant_id', 'client_id', 'client_secret')) elif service == 'azure' and candidate_kind == 'azure_container_registry': parsed = _json_object(secret_json) or {} probe = f"{parsed.get('username') or principal}:{parsed.get('password') or ''}" elif service == 'azure' and endpoint: probe = f'{endpoint}:{secret_text}' elif service == 'dockerhub' and principal: probe = f'{principal}:{secret_text}' else: probe = secret_text or secret_json probe = _bounded(probe, MAX_CANDIDATE_SECRET_BYTES) if not probe: raise ValueError('stored keycheck credential has no provider probe material') return hashlib.sha256(probe.encode('utf-8')).hexdigest() def _json_object(value): try: parsed = json.loads(str(value or '')) except (TypeError, ValueError): return None return parsed if isinstance(parsed, dict) else None def _raw_material(finding): value = finding.get('RawV2') or finding.get('Raw') if value: return str(value) structured = finding.get('StructuredData') return _json(structured) if isinstance(structured, dict) and structured else '' def _azure_foundry_keyish(value): text = re.sub(r'(?i)^bearer\s+', '', str(value or '').strip().strip('"\'`,;')).strip() lowered = text.lower() if not (20 <= len(text) <= 512) or any(character.isspace() for character in text): return False if any(marker in lowered for marker in ('http://', 'https://', '{{', '${', '<', 'azure.com')): return False if text.startswith(NON_FOUNDRY_KEY_PREFIXES): return False return bool(re.search(r'[A-Za-z]', text) and re.search(r'[0-9]', text)) def extract_azure_foundry_parts(raw, raw_v2=''): materials = [str(value or '') for value in (raw_v2, raw) if value] endpoint = next(( match.group(1).lower() for value in materials for match in [AZURE_FOUNDRY_ENDPOINT_RE.search(value)] if match ), '') if not endpoint: return None for value in materials: for key in AZURE_FOUNDRY_KEY_ASSIGNMENT_RE.findall(value): key = str(key).strip().strip('"\'`,;') if _azure_foundry_keyish(key): return {'key': key, 'endpoint': endpoint} match = AZURE_FOUNDRY_ENDPOINT_RE.search(value) if not match: continue before = value[:match.start()].strip(' \t\r\n:=,;\'"/') after = value[match.end():].strip(' \t\r\n:=,;\'"/') for key in (after, before): if _azure_foundry_keyish(key): return {'key': key, 'endpoint': endpoint} return None def extract_candidates(finding, attribution=None): if not isinstance(finding, dict): return service = _service_for_finding(finding) if not service: return detector = str(finding.get('DetectorName') or finding.get('DetectorType') or '') detector_key = re.sub(r'[^a-z0-9_]', '', detector.lower()) postman = finding.get('PostmanContext') if isinstance(finding.get('PostmanContext'), dict) else {} raw = str(finding.get('Raw') or '') raw_v2 = str(finding.get('RawV2') or '') raw_material = _raw_material(finding) base_metadata = { 'detector_name': detector, 'finding_uid': str(finding.get('finding_uid') or ''), } context = finding.get('ScannerContext') if isinstance(finding.get('ScannerContext'), dict) else {} provider_candidates = [ str(provider).lower() for provider in context.get('provider_candidates') or () if str(provider).lower() in GENERIC_SK_PROVIDERS ] provider_hint = str(context.get('provider_hint') or '') if provider_hint: base_metadata['provider_hint'] = provider_hint if provider_candidates: base_metadata['provider_candidates'] = list(dict.fromkeys(provider_candidates)) endpoint = str(postman.get('endpoint') or '') principal = str(postman.get('principal') or postman.get('username') or '') if service == 'aws': probe = raw_v2 or raw if ':' in probe: yield _make_candidate( service, 'aws_access_key_pair', probe, raw_material, secret_text=probe, metadata={**base_metadata, 'raw_v2': probe}, ) return if service == 'gcp': parsed = _json_object(raw_v2) if parsed is None: context = finding.get('ScannerContext') if isinstance(finding.get('ScannerContext'), dict) else {} parsed = _json_object(context.get('nearby')) if parsed is not None: probe = _provider_json(parsed) yield _make_candidate( service, 'gcp_json', probe, raw_material or probe, secret_json=probe, metadata={**base_metadata, 'raw_v2': probe}, ) return if service == 'azure': parsed = _json_object(raw_v2) if detector_key == 'azure': if parsed: tenant = parsed.get('tenantId') or parsed.get('tenant_id') client = parsed.get('clientId') or parsed.get('client_id') secret = parsed.get('clientSecret') or parsed.get('client_secret') if tenant and client and secret: probe = f'{tenant}:{client}:{secret}' canonical_json = _json(parsed) yield _make_candidate( service, 'azure_service_principal', probe, raw_material, secret_json=canonical_json, metadata={**base_metadata, 'raw_v2': canonical_json}, ) return if detector_key == 'azurecontainerregistry': if parsed: username = parsed.get('username') password = parsed.get('password') if username and password: probe = f'{username}:{password}' canonical_json = _json(parsed) yield _make_candidate( service, 'azure_container_registry', probe, raw_material, secret_json=canonical_json, principal=str(username), metadata={**base_metadata, 'raw_v2': canonical_json}, ) return if detector_key == 'azureopenai': match = re.match(r'^([a-fA-F0-9]{32}):(.+\.openai\.azure\.com)$', raw_v2) key = match.group(1) if match else raw found_endpoint = match.group(2).lower() if match else endpoint.lower() if key: probe = f'{found_endpoint}:{key}' if found_endpoint else key provider_raw_v2 = f'{key}:{found_endpoint}' if found_endpoint else raw_v2 yield _make_candidate( service, 'azure_openai', probe, raw_material, secret_text=key, endpoint=found_endpoint, metadata={**base_metadata, 'raw_v2': provider_raw_v2}, ) return foundry = extract_azure_foundry_parts(raw, raw_v2) if foundry: foundry_endpoint = foundry['endpoint'] foundry_key = foundry['key'] probe = f'{foundry_endpoint}:{foundry_key}' yield _make_candidate( service, 'azure_foundry', probe, raw_material, secret_text=foundry_key, endpoint=foundry_endpoint, metadata={**base_metadata, 'raw_v2': probe}, ) return if service == 'dockerhub': token_match = DOCKER_PAT_RE.search(raw_v2) or DOCKER_PAT_RE.search(raw) if not token_match: return token = token_match.group(0) username = '' if ':' in raw_v2 and raw_v2.rsplit(':', 1)[-1] == token: username = raw_v2.rsplit(':', 1)[0] extra = finding.get('ExtraData') if isinstance(finding.get('ExtraData'), dict) else {} analysis = finding.get('AnalysisInfo') if isinstance(finding.get('AnalysisInfo'), dict) else {} username = username or str(extra.get('hub_username') or analysis.get('username') or '') probe = f'{username}:{token}' if username else token yield _make_candidate( service, 'dockerhub_pat', probe, raw_material, secret_text=token, principal=username, metadata={**base_metadata, 'raw_v2': probe}, ) return patterns = { 'github': GITHUB_TOKEN_RE, 'gitlab': GITLAB_TOKEN_RE, 'gemini': GEMINI_KEY_RE, 'qwen': QWEN_KEY_RE, 'kimi': KIMI_KEY_RE, 'zai': ZAI_KEY_RE, 'provider_resolver': KIMI_KEY_RE, } pattern = patterns.get(service) values = pattern.findall(raw + '\n' + raw_v2) if pattern else [raw or raw_v2] seen = set() for value in values: value = str(value or '').strip() if not value or value in seen: continue seen.add(value) yield _make_candidate( service, 'provider_key', value, raw_material or value, secret_text=value, endpoint=endpoint, principal=principal, metadata={**base_metadata, 'raw_v2': raw_v2}, ) def extract_structured_candidates(artifact_context, attribution=None): if not isinstance(artifact_context, dict): return for finding in artifact_context.get('findings') or (): yield from extract_candidates(finding, attribution) contexts = artifact_context.get('contexts') or () origin = str(artifact_context.get('origin') or 'structured-artifact') endpoint_values = [] for context in contexts: if not isinstance(context, dict): continue text = ' '.join(str(context.get(key) or '') for key in ('value', 'endpoint', 'host', 'key')) endpoint_values.extend(AZURE_OPENAI_ENDPOINT_RE.findall(text)) endpoint_values.extend(AZURE_FOUNDRY_ENDPOINT_RE.findall(text)) seen = set() for index, context in enumerate(contexts): if not isinstance(context, dict): continue value = str(context.get('value') or '').strip() if not value or '{{' in value or '${' in value: continue path = str(context.get('path') or '') text = ' '.join(( value, str(context.get('endpoint') or ''), str(context.get('host') or ''), str(context.get('key') or ''), )) for key in GEMINI_KEY_RE.findall(value): identity = ('gemini', key) if identity in seen: continue seen.add(identity) yield _make_candidate( 'gemini', 'structured_postman', key, key, secret_text=key, metadata={ 'detector_name': 'GoogleAIStudio', 'origin': origin, 'structured_origin': f'{origin}:{path or index}', }, ) azure_key = AZURE_OPENAI_KEY_RE.search(value) if azure_key: endpoints = AZURE_OPENAI_ENDPOINT_RE.findall(text) or [ endpoint for endpoint in endpoint_values if AZURE_OPENAI_ENDPOINT_RE.fullmatch(endpoint) ] for endpoint in endpoints[:5]: key = azure_key.group(0) secret = f'{key}:{str(endpoint).lower()}' identity = ('azure-openai', secret) if identity in seen: continue seen.add(identity) yield _make_candidate( 'azure', 'structured_postman', f'{str(endpoint).lower()}:{key}', secret, secret_text=key, endpoint=str(endpoint).lower(), metadata={ 'detector_name': 'AzureOpenAI', 'origin': origin, 'raw_v2': secret, 'structured_origin': f'{origin}:{path or index}', }, ) key_label = str(context.get('key') or '').lower() normalized_key_label = re.sub(r'[^a-z0-9]+', '_', key_label).strip('_') structured_provider = None structured_detector = '' structured_pattern = None if normalized_key_label in ('dashscope_api_key', 'qwen_api_key'): structured_provider = 'qwen' structured_detector = 'QwenDashScope' structured_pattern = QWEN_KEY_RE elif normalized_key_label in ('moonshot_api_key', 'kimi_api_key'): structured_provider = 'kimi' structured_detector = 'KimiMoonshot' structured_pattern = KIMI_KEY_RE elif normalized_key_label in ( 'zai_api_key', 'z_ai_api_key', 'glm_api_key', 'zhipuai_api_key', 'bigmodel_api_key', ): structured_provider = 'zai' structured_detector = 'ZaiGLM' structured_pattern = ZAI_KEY_RE if structured_provider and structured_pattern.fullmatch(value): identity = (structured_provider, value) if identity not in seen: seen.add(identity) yield _make_candidate( structured_provider, 'structured_postman', value, value, secret_text=value, metadata={ 'detector_name': structured_detector, 'origin': origin, 'structured_origin': f'{origin}:{path or index}', }, ) foundry_endpoints = AZURE_FOUNDRY_ENDPOINT_RE.findall(text) if ( foundry_endpoints and 20 <= len(value) <= 512 and not any(character.isspace() for character in value) and any(token in key_label for token in ('key', 'token', 'secret', 'authorization')) ): for endpoint in foundry_endpoints[:5]: secret = f'{str(endpoint).lower()}:{value}' identity = ('azure-foundry', secret) if identity in seen: continue seen.add(identity) yield _make_candidate( 'azure', 'structured_postman', secret, secret, secret_text=value, endpoint=str(endpoint).lower(), metadata={ 'detector_name': 'AzureFoundryEndpointBeforeKey', 'origin': origin, 'raw_v2': secret, 'structured_origin': f'{origin}:{path or index}', }, ) def candidate_uid(scan_event_id, finding_uid_or_origin, service, credential_hash): return hashlib.sha256('|'.join(( 'truf-keycheck-candidate-v1', str(scan_event_id), str(finding_uid_or_origin), str(service), str(credential_hash), )).encode('utf-8')).hexdigest()