Skip to content
Merged
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
17 changes: 13 additions & 4 deletions src/shieldcommit/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from .scanner import scan_files
from .installer import install_hook, uninstall_hook


def get_staged_files():
"""Return list of staged files' paths."""
res = subprocess.run(["git", "diff", "--cached", "--name-only"], capture_output=True, text=True)
Expand All @@ -15,11 +16,13 @@ def get_staged_files():
return []
return [p for p in out.splitlines() if p.strip()]


@click.group()
def cli():
"""ShieldCommit - Secret Scanner for Git Commits"""
pass


@cli.command()
@click.argument("paths", nargs=-1, type=click.Path(exists=False))
def scan(paths):
Expand All @@ -45,13 +48,15 @@ def scan(paths):
else:
to_scan = get_staged_files()
if not to_scan:
click.echo("No staged files. Use `shieldcommit scan <paths>` to scan files or set staged files.")
click.echo(
"No staged files. Use `shieldcommit scan <paths>` to scan files or set staged files."
)
sys.exit(0)

result = scan_files(to_scan)
findings = result["findings"]
warnings = result["warnings"]

# Display warnings (non-blocking)
if warnings:
click.echo("⚠️ VERSION WARNINGS (Info only - no block):\n")
Expand All @@ -60,7 +65,7 @@ def scan(paths):
click.echo(f" {w['message']}")
click.echo(f" Snippet: {w['snippet']}")
click.echo("")

# Display findings (blocking)
if not findings:
click.echo("✓ No secrets found.")
Expand All @@ -73,9 +78,12 @@ def scan(paths):
click.echo(f" Confidence: {f.get('confidence', 'N/A'):.2%}")
click.echo(f" Snippet: {f['snippet']}")
click.echo("")
click.echo("Your commit or action has been blocked. Remove or rotate secrets before proceeding.")
click.echo(
"Your commit or action has been blocked. Remove or rotate secrets before proceeding."
)
sys.exit(1)


@cli.command()
def install():
"""Install pre-commit hook in current repo."""
Expand All @@ -85,6 +93,7 @@ def install():
else:
click.echo("❌ Failed to install hook.")


@cli.command()
def uninstall():
ok = uninstall_hook(".")
Expand Down
91 changes: 51 additions & 40 deletions src/shieldcommit/aks_detector.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
"1.22": {"status": "deprecated", "eol": "2023-11"},
}


def parse_terraform_aks_versions(content: str) -> List[Dict[str, Any]]:
"""
Parse Terraform file content for AKS cluster versions.
Expand All @@ -31,95 +32,105 @@ def parse_terraform_aks_versions(content: str) -> List[Dict[str, Any]]:
"""
findings = []
lines = content.splitlines()

# Pattern 1: Direct resource usage: kubernetes_version = "1.27"
pattern1 = r'kubernetes_version\s*=\s*["\']([0-9]+\.[0-9]+)["\']'

# Pattern 2: Variable default values: default = "1.27" (for cluster_version or similar)
pattern2 = r'default\s*=\s*["\']([0-9]+\.[0-9]+)["\']'

in_cluster_version_var = False

for line_no, line in enumerate(lines, 1):
# Track if we're in a cluster_version variable block
if 'variable' in line and 'cluster_version' in line:
if "variable" in line and "cluster_version" in line:
in_cluster_version_var = True
elif in_cluster_version_var and '}' in line:
elif in_cluster_version_var and "}" in line:
in_cluster_version_var = False

# Check pattern 1: Direct kubernetes_version
matches = re.finditer(pattern1, line)
for match in matches:
version = match.group(1)
findings.append({
"line": line_no,
"version": version,
"snippet": line.strip(),
"match": match.group(0)
})

findings.append(
{
"line": line_no,
"version": version,
"snippet": line.strip(),
"match": match.group(0),
}
)

# Check pattern 2: Variable default (when in cluster_version variable block)
if in_cluster_version_var:
matches = re.finditer(pattern2, line)
for match in matches:
version = match.group(1)
findings.append({
"line": line_no,
"version": version,
"snippet": line.strip(),
"match": match.group(0)
})

findings.append(
{
"line": line_no,
"version": version,
"snippet": line.strip(),
"match": match.group(0),
}
)

return findings


def get_version_warning(version: str) -> str:
"""Get warning message for a given AKS version."""
if version not in AKS_VERSIONS:
return f"⚠️ AKS version {version} is unknown. Check Azure documentation."

info = AKS_VERSIONS[version]
status = info["status"]
eol = info["eol"]

if status == "deprecated":
return f"🚨 AKS {version} is DEPRECATED (EOL: {eol}). Upgrade immediately."
elif status == "extended":
return f"⚠️ AKS {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading."
return (
f"⚠️ AKS {version} on Extended Support (EOL: {eol}). Higher costs. Consider upgrading."
)
else: # current
return f"✓ AKS {version} is currently supported (EOL: {eol})."


def scan_aks_versions(file_path: Path) -> List[Dict[str, Any]]:
"""
Scan a Terraform file for AKS version warnings.
Returns list of warnings.
"""
warnings = []

# Only scan Terraform files
if file_path.suffix not in {'.tf', '.json'}:
if file_path.suffix not in {".tf", ".json"}:
return warnings

try:
content = file_path.read_text(errors="ignore")
except Exception:
return warnings

findings = parse_terraform_aks_versions(content)

for finding in findings:
version = finding["version"]
warning_msg = get_version_warning(version)

# Only warn on non-current versions
if version in AKS_VERSIONS and AKS_VERSIONS[version]["status"] != "current":
warnings.append({
"file": str(file_path),
"line": finding["line"],
"type": "aks_version",
"version": version,
"status": AKS_VERSIONS[version]["status"],
"message": warning_msg,
"snippet": finding["snippet"]
})

warnings.append(
{
"file": str(file_path),
"line": finding["line"],
"type": "aks_version",
"version": version,
"status": AKS_VERSIONS[version]["status"],
"message": warning_msg,
"snippet": finding["snippet"],
}
)

return warnings
Loading
Loading