feat: consolidate legacy workflows into gyxx-flow
This commit is contained in:
@@ -0,0 +1,5 @@
|
||||
"""Security checks used by migration and release gates."""
|
||||
|
||||
from gyxx_flow.security.scanner import SecretFinding, scan_repository
|
||||
|
||||
__all__ = ["SecretFinding", "scan_repository"]
|
||||
@@ -0,0 +1,174 @@
|
||||
"""Find high-confidence plaintext credentials without exposing their values."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
DEFAULT_EXCLUDED_DIRECTORIES = frozenset({"var", ".git", ".venv", ".learnings"})
|
||||
|
||||
_ASSIGNMENT_PATTERN = re.compile(
|
||||
r"""
|
||||
[\"']?
|
||||
(?:[a-z0-9]+[_\-.])*
|
||||
(?:
|
||||
password | passwd | pwd |
|
||||
api[_-]?key | app[_-]?secret | client[_-]?secret | consumer[_-]?secret |
|
||||
secret[_-]?key |
|
||||
secret[_-]?access[_-]?key | access[_-]?key(?:[_-]?id)? |
|
||||
access[_-]?token | refresh[_-]?token | auth[_-]?token
|
||||
)
|
||||
[\"']?
|
||||
\s*(?::|(?<![=!<>])=(?!=))\s*
|
||||
(?:
|
||||
\"(?P<double>[^\"\r\n]*)\" |
|
||||
'(?P<single>[^'\r\n]*)' |
|
||||
(?P<bare>[^\s#;,]+)
|
||||
)
|
||||
""",
|
||||
re.IGNORECASE | re.VERBOSE,
|
||||
)
|
||||
_CREDENTIAL_URL_PATTERN = re.compile(
|
||||
r"\b(?:https?|ftp)://[^\s/:@]+:[^\s/@{}$%]+@[^\s/\"']+",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_PRIVATE_KEY_PATTERN = re.compile(
|
||||
r"-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----"
|
||||
)
|
||||
_NAMED_LONG_HEX_CREDENTIAL_PATTERN = re.compile(
|
||||
r"""
|
||||
(?:api[_-]?key|app[_-]?secret|client[_-]?secret|access[_-]?token)
|
||||
\s*=\s*
|
||||
(?:os\.getenv\([\s\S]{0,160}?,\s*)?
|
||||
["'][0-9a-f]{32,}["']
|
||||
""",
|
||||
re.IGNORECASE | re.VERBOSE,
|
||||
)
|
||||
|
||||
_PLACEHOLDER_WORDS = frozenset(
|
||||
{
|
||||
"changeme",
|
||||
"example",
|
||||
"none",
|
||||
"null",
|
||||
"placeholder",
|
||||
"redacted",
|
||||
"replaceme",
|
||||
"secret",
|
||||
"secretstr",
|
||||
"str",
|
||||
"string",
|
||||
"todo",
|
||||
"xxx",
|
||||
"yourkey",
|
||||
"yourpassword",
|
||||
"yoursecret",
|
||||
"yourtoken",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SecretFinding:
|
||||
"""A deliberately value-free secret scan result."""
|
||||
|
||||
file: str
|
||||
line: int
|
||||
rule: str
|
||||
|
||||
|
||||
def scan_repository(
|
||||
root: str | Path,
|
||||
*,
|
||||
excluded_directories: frozenset[str] = DEFAULT_EXCLUDED_DIRECTORIES,
|
||||
) -> list[SecretFinding]:
|
||||
"""Scan repository text files and return deterministic, value-free findings.
|
||||
|
||||
Directory names in ``excluded_directories`` are pruned at every depth. Binary
|
||||
and undecodable files are ignored. The function never stores matched values in
|
||||
its result objects.
|
||||
"""
|
||||
|
||||
repository = Path(root).expanduser().resolve()
|
||||
if not repository.is_dir():
|
||||
raise NotADirectoryError(repository)
|
||||
|
||||
findings: list[SecretFinding] = []
|
||||
for directory, directory_names, file_names in os.walk(repository, followlinks=False):
|
||||
directory_names[:] = sorted(
|
||||
name for name in directory_names if name not in excluded_directories
|
||||
)
|
||||
current_directory = Path(directory)
|
||||
for file_name in sorted(file_names):
|
||||
path = current_directory / file_name
|
||||
text = _read_text(path)
|
||||
if text is None:
|
||||
continue
|
||||
relative_file = path.relative_to(repository).as_posix()
|
||||
findings.extend(_scan_text(relative_file, text))
|
||||
return findings
|
||||
|
||||
|
||||
def _read_text(path: Path) -> str | None:
|
||||
try:
|
||||
content = path.read_bytes()
|
||||
except OSError:
|
||||
return None
|
||||
if b"\x00" in content:
|
||||
return None
|
||||
for encoding in ("utf-8-sig", "gb18030"):
|
||||
try:
|
||||
return content.decode(encoding)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def _scan_text(relative_file: str, text: str) -> list[SecretFinding]:
|
||||
findings: list[SecretFinding] = []
|
||||
if relative_file.casefold().endswith((".py", ".pyw")):
|
||||
for match in _NAMED_LONG_HEX_CREDENTIAL_PATTERN.finditer(text):
|
||||
findings.append(
|
||||
SecretFinding(
|
||||
relative_file,
|
||||
text.count("\n", 0, match.start()) + 1,
|
||||
"hardcoded-long-hex-credential",
|
||||
)
|
||||
)
|
||||
for line_number, line in enumerate(text.splitlines(), start=1):
|
||||
if _PRIVATE_KEY_PATTERN.search(line):
|
||||
findings.append(
|
||||
SecretFinding(relative_file, line_number, "private-key-material")
|
||||
)
|
||||
if _CREDENTIAL_URL_PATTERN.search(line):
|
||||
findings.append(SecretFinding(relative_file, line_number, "credential-in-url"))
|
||||
if any(_is_plaintext_value(match) for match in _ASSIGNMENT_PATTERN.finditer(line)):
|
||||
findings.append(
|
||||
SecretFinding(relative_file, line_number, "plaintext-credential")
|
||||
)
|
||||
return findings
|
||||
|
||||
|
||||
def _is_plaintext_value(match: re.Match[str]) -> bool:
|
||||
value = next(
|
||||
(candidate for candidate in match.group("double", "single", "bare") if candidate is not None),
|
||||
"",
|
||||
).strip()
|
||||
if not value:
|
||||
return False
|
||||
|
||||
lowered = value.casefold()
|
||||
compact = re.sub(r"[^a-z0-9]", "", lowered)
|
||||
if compact in _PLACEHOLDER_WORDS:
|
||||
return False
|
||||
if lowered in {"true", "false"}:
|
||||
return False
|
||||
if value.startswith(("$", "%(", "{{", "{")):
|
||||
return False
|
||||
if re.fullmatch(r"[a-z_]\w*(?:\.[a-z_]\w*)+", lowered):
|
||||
return False
|
||||
if any(marker in lowered for marker in ("getenv(", "environ[", "secretmanager", "keyvault")):
|
||||
return False
|
||||
return True
|
||||
Reference in New Issue
Block a user