Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 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
104 changes: 97 additions & 7 deletions socket_basics/core/connector/trufflehog/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,10 @@

import json
import logging
import re
import subprocess
import os
import tempfile
from typing import Dict, List, Any

from ..base import BaseConnector
Expand All @@ -31,6 +33,83 @@ def is_enabled(self) -> bool:
"""Check if secret scanning should be enabled"""
return self.config.get('secret_scanning_enabled', False)

@staticmethod
def _path_regex(value: str) -> str:
"""Escape a filesystem path for TruffleHog's Go regular-expression input."""
normalized = str(value).replace('\\', '/')
return re.escape(normalized).replace('/', r'[/\\]')

def _workspace_root(self) -> str:
"""Return the absolute workspace path used by the scanner command."""
workspace = getattr(self.config, 'workspace', None)
if not isinstance(workspace, (str, bytes, os.PathLike)):
workspace = (
getattr(workspace, 'path', None)
or getattr(workspace, 'root', None)
or workspace
)
if not workspace:
return ''
try:
return os.path.abspath(os.fspath(workspace)).rstrip('/\\')
except (TypeError, ValueError):
return ''

def _build_exclude_patterns(self, exclude_dirs: Any) -> List[str]:
"""Build workspace-relative path patterns for TruffleHog.

TruffleHog expects one --exclude-paths value containing a file of
newline-separated regular expressions. The configured values are
directory names, so anchor each one below the workspace root. This
prevents a directory such as tmp from matching the workspace's
parent path (for example /tmp/...), and prevents .git from
matching .github.
"""
if isinstance(exclude_dirs, str):
entries = exclude_dirs.split(',')
else:
entries = exclude_dirs or []

workspace_root = self._workspace_root()
patterns = []
for entry in entries:
directory = str(entry).strip().replace('\\', '/').strip('/')
if not directory:
continue

directory_regex = self._path_regex(directory)
if workspace_root:
root_regex = self._path_regex(workspace_root)
patterns.append(
rf'^{root_regex}[/\\](?:.*[/\\])?{directory_regex}(?:[/\\]|$)'
)
else:
patterns.append(rf'(?:^|[/\\]){directory_regex}(?:[/\\]|$)')

return patterns

def _write_exclude_file(self, exclude_dirs: Any) -> str | None:
"""Write exclude regexes to a temporary file for TruffleHog."""
patterns = self._build_exclude_patterns(exclude_dirs)
if not patterns:
return None

with tempfile.NamedTemporaryFile(
mode='w',
encoding='utf-8',
prefix='socket-basics-trufflehog-',
suffix='.txt',
delete=False,
) as exclude_file:
exclude_file.write('\n'.join(patterns))
exclude_file.write('\n')
return exclude_file.name

@staticmethod
def _absolute_scan_target(target: Any) -> str:
"""Return the absolute path string TruffleHog will filter against."""
return os.path.abspath(os.fspath(target))

def scan(self) -> Dict[str, Any]:
"""Run Trufflehog secret scanning"""
if not self.is_enabled():
Expand All @@ -41,7 +120,8 @@ def scan(self) -> Dict[str, Any]:

targets = self.config.get_scan_targets()
results = {}


exclude_file_path = None
try:
# Prefer explicit changed_files, fallback to git staged
changed_files = self.config.get('changed_files', []) if hasattr(self.config, '_config') else []
Expand All @@ -59,18 +139,20 @@ def scan(self) -> Dict[str, Any]:
'--no-verification' if not self.config.get('trufflehog_show_unverified', False) else '--include-detectors=all'
]

# Add exclusion patterns
# TruffleHog accepts --exclude-paths only once and expects a file
# containing newline-separated regular expressions.
exclude_dirs = self.config.get('trufflehog_exclude_dir', '')
if exclude_dirs:
for exclude_dir in exclude_dirs.split(','):
cmd.extend(['--exclude-paths', exclude_dir.strip()])
exclude_file_path = self._write_exclude_file(exclude_dirs)
if exclude_file_path:
cmd.extend(['--exclude-paths', exclude_file_path])

# If changed_files present, pass those individual files, otherwise use configured targets
if changed_files:
for cf in changed_files:
cmd.append(str(self.config.workspace / cf))
cmd.append(self._absolute_scan_target(self.config.workspace / cf))
else:
cmd.extend(targets)
cmd.extend(self._absolute_scan_target(target) for target in targets)

logger.info(f"Running: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True)
Expand Down Expand Up @@ -138,7 +220,15 @@ def scan(self) -> Dict[str, Any]:
logger.error("Trufflehog not found. Please install Trufflehog")
except Exception as e:
logger.error(f"Error running Trufflehog: {e}")

finally:
if exclude_file_path:
try:
os.unlink(exclude_file_path)
except FileNotFoundError:
pass
except OSError as e:
logger.warning(f"Failed to remove Trufflehog exclude file: {e}")

return results

def _convert_to_socket_facts(self, raw_results: Any) -> Dict[str, Any]:
Expand Down
119 changes: 119 additions & 0 deletions tests/test_trufflehog_excludes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
from pathlib import Path
import re
from types import SimpleNamespace

from socket_basics.core.connector.trufflehog import TruffleHogScanner


def _scanner(tmp_path, exclude_dirs):
config = SimpleNamespace(
workspace=tmp_path,
trufflehog_exclude_dir=exclude_dirs,
)
scanner = TruffleHogScanner.__new__(TruffleHogScanner)
scanner.config = config
return scanner


def test_build_exclude_patterns_are_anchored_to_workspace(tmp_path):
scanner = _scanner(tmp_path, "")

patterns = scanner._build_exclude_patterns("node_modules,.git,tmp")

assert len(patterns) == 3
workspace_regex = scanner._path_regex(str(tmp_path))
assert all(workspace_regex in pattern for pattern in patterns)
assert any(r"\.git" in pattern for pattern in patterns)
assert all(".github" not in pattern for pattern in patterns)
assert all(not pattern.startswith(r"(?:^|[/\\])") for pattern in patterns)

def matches(path):
return any(re.search(pattern, path) for pattern in patterns)

assert matches(str(tmp_path / "src" / "node_modules" / "package.json"))
assert not matches(str(tmp_path / ".github" / "workflows" / "scan.yml"))
assert not re.search(patterns[2], str(tmp_path / "app.py"))


def test_build_exclude_patterns_skip_empty_entries(tmp_path):
scanner = _scanner(tmp_path, "")

patterns = scanner._build_exclude_patterns(" node_modules, ,dist, ")

assert len(patterns) == 2
assert all("node_modules" in pattern or "dist" in pattern for pattern in patterns)


def test_write_exclude_file_contains_one_pattern_per_line(tmp_path):
scanner = _scanner(tmp_path, "")

exclude_file = scanner._write_exclude_file("node_modules,.git")
try:
contents = Path(exclude_file).read_text(encoding="utf-8").splitlines()
finally:
Path(exclude_file).unlink()

assert len(contents) == 2
workspace_regex = scanner._path_regex(str(tmp_path))
assert all(workspace_regex in pattern for pattern in contents)


def test_scan_passes_one_exclude_paths_flag_and_cleans_up(tmp_path, monkeypatch):
scanner = _scanner(tmp_path, "node_modules,.yarn,dist")
scanner.is_enabled = lambda: True
scanner.config.get = lambda key, default=None: {
"trufflehog_exclude_dir": scanner.config.trufflehog_exclude_dir,
"trufflehog_show_unverified": False,
}.get(key, default)
scanner.config.get_scan_targets = lambda: [str(tmp_path)]
scanner._process_results = lambda findings: {}

captured = {}

def fake_run(command, **kwargs):
captured["command"] = command
exclude_path = Path(command[command.index("--exclude-paths") + 1])
captured["contents"] = exclude_path.read_text(encoding="utf-8").splitlines()
captured["exists_during_run"] = exclude_path.exists()
return SimpleNamespace(returncode=0, stdout="", stderr="")

monkeypatch.setattr(
"socket_basics.core.connector.trufflehog.subprocess.run",
fake_run,
)

scanner.scan()

command = captured["command"]
assert command.count("--exclude-paths") == 1
assert captured["exists_during_run"] is True
assert len(captured["contents"]) == 3
assert not Path(command[command.index("--exclude-paths") + 1]).exists()


def test_scan_uses_absolute_targets_for_relative_workspace(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
scanner = _scanner(Path("."), "node_modules")
scanner.is_enabled = lambda: True
scanner.config.get = lambda key, default=None: {
"trufflehog_exclude_dir": scanner.config.trufflehog_exclude_dir,
"trufflehog_show_unverified": False,
}.get(key, default)
scanner.config.get_scan_targets = lambda: ["."]
scanner._process_results = lambda findings: {}

captured = {}

def fake_run(command, **kwargs):
captured["command"] = command
return SimpleNamespace(returncode=0, stdout="", stderr="")

monkeypatch.setattr(
"socket_basics.core.connector.trufflehog.subprocess.run",
fake_run,
)

scanner.scan()

command = captured["command"]
assert command[-1] == str(tmp_path)