Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

API for python scripts Issue #713 #836

Open
wants to merge 12 commits into
base: master
Choose a base branch
from
Prev Previous commit
Next Next commit
Refactored API api.py
  • Loading branch information
issabayevmk authored Jun 7, 2024
commit f7862ddc57f21c010f099a96977cb1b494ae29f9
233 changes: 119 additions & 114 deletions detect_secrets/api.py
Original file line number Diff line number Diff line change
@@ -1,148 +1,152 @@
import os
import importlib
import pkgutil
import inspect
from abc import ABC
from git import Repo, InvalidGitRepositoryError
from detect_secrets import SecretsCollection
from detect_secrets.settings import default_settings, transient_settings

from detect_secrets import plugins as ds_plugins
from detect_secrets.plugins.base import BasePlugin


def is_concrete_class(cls):
return (
not inspect.isabstract(cls)
and issubclass(cls, BasePlugin)
and cls is not BasePlugin
)


def load_all_plugins():
def get_settings(filters=None, plugins=None):
"""
Load and return all available plugins from detect-secrets.
Return used plugins and filters to be used to scan with provided params
"""
plugins = []
package = ds_plugins
for _, module_name, _ in pkgutil.iter_modules(package.__path__):
module = importlib.import_module(f"{package.__name__}.{module_name}")
for name, obj in inspect.getmembers(module, inspect.isclass):
if is_concrete_class(obj):
plugins.append(obj())
return plugins
if filters and not isinstance(filters, list):
raise ValueError(f"Error: 'filters' must be List object")

if plugins and not isinstance(plugins, list):
raise ValueError(f"Error: 'plugins' must be List object")

def load_plugin_by_name(plugin_name: str):
"""
Dynamically load and return an instance of the specified plugin by name.
"""
package = ds_plugins
for _, module_name, _ in pkgutil.iter_modules(package.__path__):
module = importlib.import_module(f"{package.__name__}.{module_name}")
for name, obj in inspect.getmembers(module, inspect.isclass):
if name == plugin_name and is_concrete_class(obj):
return obj()
raise ValueError(
f"Error: no plugin found with name: '{plugin_name}'. To get the list of supported plugins, call list_plugins()"
)


def load_specified_plugins(plugin_names: [str]):
"""
Dynamically load and return specified plugins by name.
"""
plugins = []
for plugin_name in plugin_names:
plugins.append(load_plugin_by_name(plugin_name))
return plugins
if filters:
filters_used = filters
else:
filters_used = []
with default_settings() as settings:
for key in settings.filters:
filters_used.append({'path': key})

if plugins:
plugins_used = plugins
else:
plugins_used = []
with default_settings() as settings:
for key in settings.plugins:
plugins_used.append({'name': key})

def list_plugins():
"""
Retunr a list of available plugins to use.
"""
plugins = []
package = ds_plugins
for _, module_name, _ in pkgutil.iter_modules(package.__path__):
module = importlib.import_module(f"{package.__name__}.{module_name}")
for name, obj in inspect.getmembers(module, inspect.isclass):
if is_concrete_class(obj):
plugins.append(name)
return plugins
return {"plugins": plugins_used, "filters": filters_used}


def scan_string(string_to_check: str, plugins: str = "all"):
def scan_string(string: str, filters=None, plugins=None):
"""
Scan a string for secrets using the specified plugins.
Scan a string for secrets using detect-secrets with custom filters and plugins

Args:
string_to_check (str): string to to scan for secrets.
plugins (str): Names of the comma (,) separated detect-secrets plugin names to use.
:param string: String to scan
:param filters: Custom filters for detect-secrets
:param plugins: Custom plugins for detect-secrets
:return: Detected secrets in str format
"""
if not isinstance(string_to_check, str):
raise ValueError(f"Error: '{string_to_check}' must be 'string' object")
if not isinstance(string, str):
raise ValueError(f"Error: '{string}' must be 'string' formatted path to a file")

if filters and not isinstance(filters, list):
raise ValueError(f"Error: 'filters' must be List object")

if plugins and not isinstance(plugins, list):
raise ValueError(f"Error: 'plugins' must be List object")

# Initialize a SecretsCollection
secrets = SecretsCollection()

# Load default settings if no filters and plugins provided:
if not filters and not plugins:
settings = default_settings()
# Scan the string
with settings:
secrets.scan_string(string)
return secrets.json()
elif filters and not plugins:
filters_used = filters
plugins_used = get_settings(plugins=plugins).get("plugins")
elif not filters and plugins:
plugins_used = plugins
filters_used = get_settings(filters=filters).get("filters")
else:
filters_used = filters
plugins_used = plugins
# Scan the string
with transient_settings(
{"plugins_used": plugins_used, "filters_used": filters_used}
) as settings:
secrets.scan_string(string)
return secrets.json()

if not isinstance(plugins, str):
raise ValueError(
f"Error: '{plugins}' must be comma (,) sepated 'string' object"
)

if plugins == "all":
detectors = load_all_plugins()
else:
plugin_names = plugins.split(",")
detectors = load_specified_plugins(plugin_names)

found_secrets = {}
for detector in detectors:
secrets = detector.analyze_string(string_to_check)
detector_name = detector.json().get("name")
for secret in secrets:
if detector_name not in found_secrets:
found_secrets[detector_name] = [secret]
elif secret not in found_secrets[detector_name]:
found_secrets[detector_name].append(secret)
return found_secrets


def scan_file(filepath: str, plugins: str = "all"):
def scan_file(filepath, filters=None, plugins=None):
"""
Scan a local file for secrets using the specified plugins.
Scan a file for secrets using detect-secrets with custom filters and plugins

Args:
filepath (str): Path to the local file.
plugins (str): Names of the comma (,) separated detect-secrets plugin names to use.
:param filepath: Path to the file to scan
:param filters: Custom filters for detect-secrets
:param plugins: Custom plugins for detect-secrets
:return: Detected secrets in str format
"""
if not isinstance(filepath, str):
raise ValueError(
f"Error: '{filepath}' must be 'string' formatted path to a file"
)

if filters and not isinstance(filters, list):
raise ValueError(f"Error: 'filters' must be List object")

if plugins and not isinstance(plugins, list):
raise ValueError(f"Error: 'plugins' must be List object")

try:
with open(filepath, "r") as file:
lines = file.readlines()
found_secrets = {}
for idx, line in enumerate(lines):
secrets_in_line = scan_string(line, plugins)
if secrets_in_line != {}:
found_secrets[f"Line {idx + 1}"] = secrets_in_line
return found_secrets
with open(filepath, "r") as f:
f.read()
except Exception as e:
raise ValueError(f"Error scanning '{filepath}': {e}")
return e
# Initialize a SecretsCollection
secrets = SecretsCollection()

# Load default settings if no filters and plugins provided:
if not filters and not plugins:
settings = default_settings()
# Scan the file
with settings:
secrets.scan_file(filepath)
return secrets.json()
elif filters and not plugins:
filters_used = filters
plugins_used = get_settings(plugins=plugins).get("plugins")
elif not filters and plugins:
plugins_used = plugins
filters_used = get_settings(filters=filters).get("filters")
else:
filters_used = filters
plugins_used = plugins

# Scan a file
with transient_settings(
{"plugins_used": plugins_used, "filters_used": filters_used}
) as settings:
secrets.scan_file(filepath)
return secrets.json()


def scan_git_repository(
repo_path: str, plugins: str = "all", scan_all_files: bool = False
repo_path: str, plugins=None, filters=None, scan_all_files: bool = False
):
"""
Scan a local Git repository for secrets using the specified plugins.
Scan a local Git repository for secrets using the specified plugins and filters

Args:
repo_path (str): Path to the local Git repository.
plugins (str): Names of the comma (,) separated detect-secrets plugin names to use.
scan_all_files (bool): If True, scan all files in the repository. If False, scan only Git-tracked files.
:param repo_path: Path to the local Git repository
:param filters: Custom filters for detect-secrets
:param plugins: Custom plugins for detect-secrets
:param scan_all_files (bool): If True, scan all files in the repository. If False, scan only Git-tracked files.
:return: Detected secrets in List format
"""
if not isinstance(scan_all_files, bool):
raise ValueError(f"Error: 'scan_all_files' must be 'bool' type")
if not isinstance(repo_path, str):
raise ValueError(f"Error: 'repo_path' must be 'str' type path to repository")

try:
repo = Repo(repo_path)
Expand All @@ -164,11 +168,12 @@ def scan_git_repository(
[os.path.join(repo_path, item) for item in repo.untracked_files]
)

found_secrets = {}
results = []
for filepath in files_to_scan:
secrets_in_file = scan_file(filepath, plugins)
if secrets_in_file != {}:
found_secrets[filepath] = secrets_in_file
return found_secrets
secrets = scan_file(filepath, plugins=plugins, filters=filters)
if secrets != {}:
results.append(secrets)
return results

except InvalidGitRepositoryError:
raise ValueError(f"Error: '{repo_path}' is not a valid Git repositoty")