Handle non-interactive shells gracefully (#158)

Author: ikretz

README.md

@@ -93,6 +93,16 @@ $ scfw run poetry add git+https://github.com/DataDog/guarddog
 
 For `pip install` commands, packages will be installed in the same environment (virtual or global) in which the command was run.
 
+Several command-line options of the `run` subcommand are noteworthy:
+
+* `--dry-run`: Verify any installation targets but do not run the package manager command. The exit code indicates whether there were findings of any severity
+
+* `--allow-on-warning` and `--block-on-warning`: Non-interactively allow or block commands, respectively, with only warning-level findings. Setting the environment variable `SCFW_ON_WARNING` to `"ALLOW"` or `"BLOCK"` achieves the same effect, with the CLI options taking priority over the environment variable when both are used
+
+* `--error-on-block`: Treat blocked commands as errors (useful for scripting)
+
+Run `scfw run --help` to see all available command-line options.
+
 ### Audit installed packages
 
 Supply-Chain Firewall can also use its verifiers to audit installed packages:

scfw/configure/__init__.py

@@ -4,7 +4,6 @@
 
 from argparse import Namespace
 
-from scfw.configure.constants import *  # noqa
 import scfw.configure.dd_agent as dd_agent
 import scfw.configure.env as env
 import scfw.configure.interactive as interactive

scfw/configure/dd_agent.py

@@ -3,11 +3,14 @@
 """
 
 import json
+import logging
 from pathlib import Path
 import shutil
 import subprocess
 
-from scfw.configure.constants import DD_SERVICE, DD_SOURCE
+from scfw.constants import DD_SERVICE, DD_SOURCE
+
+_log = logging.getLogger(__name__)
 
 
 def configure_agent_logging(port: str):
@@ -21,7 +24,7 @@ def configure_agent_logging(port: str):
         ValueError: An invalid port number was provided.
         RuntimeError: An error occurred while querying the Agent's status.
     """
-    if not (0 < int(port) < 65536):
+    if not (0 < int(port) < 2 ** 16):
         raise ValueError("Invalid port number provided for Datadog Agent logging")
 
     config_file = (
@@ -37,8 +40,10 @@ def configure_agent_logging(port: str):
 
     if not scfw_config_dir.is_dir():
         scfw_config_dir.mkdir()
+        _log.info(f"Created directory {scfw_config_dir} for Datadog Agent configuration")
     with open(scfw_config_file, 'w') as f:
         f.write(config_file)
+        _log.info(f"Wrote file {scfw_config_file} with Datadog Agent configuration")
 
 
 def remove_agent_logging():
@@ -51,13 +56,15 @@ def remove_agent_logging():
     scfw_config_dir = _dd_agent_scfw_config_dir()
 
     if not scfw_config_dir.is_dir():
+        _log.info("No Datadog Agent configuration directory to remove")
         return
 
     try:
         shutil.rmtree(scfw_config_dir)
+        _log.info(f"Deleted directory {scfw_config_dir} with Datadog Agent configuration")
     except Exception:
         raise RuntimeError(
-            "Failed to delete Datadog Agent configuration directory for Supply-Chain Firewall"
+            f"Failed to delete directory {scfw_config_dir} with Datadog Agent configuration for Supply-Chain Firewall"
         )
 
 
@@ -71,19 +78,30 @@ def _dd_agent_scfw_config_dir() -> Path:
 
     Raises:
         RuntimeError:
-            Unable to query Datadog Agent status to read the location of its
-            global configuration directory.
+            * Unable to query Datadog Agent status to read the location of its global
+              configuration directory
+            * Datadog Agent global configuration directory is not set or does not exist
+        ValueError: Failed to parse Datadog Agent status JSON report.
     """
     try:
         agent_status = subprocess.run(
             ["datadog-agent", "status", "--json"], check=True, text=True, capture_output=True
         )
-        agent_config_dir = json.loads(agent_status.stdout).get("config", {}).get("confd_path", "")
+        config_confd_path = json.loads(agent_status.stdout).get("config", {}).get("confd_path")
+        agent_config_dir = Path(config_confd_path) if config_confd_path else None
 
     except subprocess.CalledProcessError:
         raise RuntimeError(
             "Unable to query Datadog Agent status: please ensure the Agent is running. "
             "Linux users may need sudo to run this command."
         )
 
-    return Path(agent_config_dir) / "scfw.d"
+    except json.JSONDecodeError:
+        raise ValueError("Failed to parse Datadog Agent status report as JSON")
+
+    if not (agent_config_dir and agent_config_dir.is_dir()):
+        raise RuntimeError(
+            "Datadog Agent global configuration directory is not set or does not exist"
+        )
+
+    return agent_config_dir / "scfw.d"

scfw/configure/env.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 import re
 
-from scfw.configure.constants import DD_AGENT_PORT_VAR, DD_API_KEY_VAR, DD_LOG_LEVEL_VAR, SCFW_HOME_VAR
+from scfw.constants import DD_AGENT_PORT_VAR, DD_API_KEY_VAR, DD_LOG_LEVEL_VAR, SCFW_HOME_VAR
 
 _log = logging.getLogger(__name__)
 

scfw/configure/interactive.py

@@ -9,7 +9,7 @@
 
 import inquirer  # type: ignore
 
-from scfw.configure.constants import DD_API_KEY_VAR
+from scfw.constants import DD_API_KEY_VAR
 from scfw.logger import FirewallAction
 
 GREETING = (

scfw/configure/constants.py renamed to scfw/constants.py

@@ -33,6 +33,12 @@
 forward firewall logs to the local Datadog Agent.
 """
 
+ON_WARNING_VAR = "SCFW_ON_WARNING"
+"""
+The environment variable under which the firewall looks for the user's choice of
+`FirewallAction` to take for commands with only warning-level findings.
+"""
+
 SCFW_HOME_VAR = "SCFW_HOME"
 """
 The environment variable under which the firewall looks for its home (cache) directory.

scfw/firewall.py

@@ -5,7 +5,10 @@
 from argparse import Namespace
 import inquirer  # type: ignore
 import logging
+import os
+import sys
 
+from scfw.constants import ON_WARNING_VAR
 from scfw.logger import FirewallAction
 from scfw.loggers import FirewallLoggers
 from scfw.package_manager import UnsupportedVersionError
@@ -65,10 +68,7 @@ def run_firewall(args: Namespace) -> int:
 
             if not args.dry_run and warning_report:
                 print(warning_report)
-                if (
-                    not args.allow_on_warning
-                    and (args.block_on_warning or not inquirer.confirm("Proceed with installation?", default=False))
-                ):
+                if _get_warning_action(args.allow_on_warning, args.block_on_warning) == FirewallAction.BLOCK:
                     loggers.log_firewall_action(
                         package_manager.ecosystem(),
                         package_manager.name(),
@@ -128,3 +128,38 @@ def run_firewall(args: Namespace) -> int:
             warned=False,
         )
         return package_manager.run_command(args.command)
+
+
+def _get_warning_action(cli_allow_choice: bool, cli_block_choice: bool) -> FirewallAction:
+    """
+    Return the `FirewallAction` that should be taken for `WARNING`-level findings.
+
+    Args:
+        cli_allow_choice:
+            A `bool` indicating whether the user selected `--allow-on-warning` on the command-line.
+        cli_block_choice:
+            A `bool` indicating whether the user selected `--block-on-warning` on the command-line.
+
+    Returns:
+        The `FirewallAction` that should be taken based on the user's configured choices or, if
+        no choice has been made, on the user's runtime (interactive) decision.
+    """
+    if cli_block_choice:
+        return FirewallAction.BLOCK
+    if cli_allow_choice:
+        return FirewallAction.ALLOW
+
+    if (action := os.getenv(ON_WARNING_VAR)):
+        try:
+            return FirewallAction.from_string(action)
+        except Exception:
+            _log.warning(f"Ignoring invalid firewall action {ON_WARNING_VAR}='{action}'")
+
+    if not sys.stdin.isatty():
+        _log.warning(
+            "Non-interactive terminal and no predefined action for WARNING findings: defaulting to BLOCK"
+        )
+        return FirewallAction.BLOCK
+
+    user_confirmed = inquirer.confirm("Proceed with installation?", default=False)
+    return FirewallAction.ALLOW if user_confirmed else FirewallAction.BLOCK

scfw/loggers/dd_agent_logger.py

@@ -6,7 +6,7 @@
 import os
 import socket
 
-from scfw.configure import DD_AGENT_PORT_VAR
+from scfw.constants import DD_AGENT_PORT_VAR
 from scfw.logger import FirewallLogger
 from scfw.loggers.dd_logger import DDLogFormatter, DDLogger
 

scfw/loggers/dd_api_logger.py

@@ -12,7 +12,7 @@
 from datadog_api_client.v2.model.http_log_item import HTTPLogItem
 
 import scfw
-from scfw.configure import DD_API_KEY_VAR, DD_ENV, DD_SERVICE, DD_SOURCE
+from scfw.constants import DD_API_KEY_VAR, DD_ENV, DD_SERVICE, DD_SOURCE
 from scfw.logger import FirewallLogger
 from scfw.loggers.dd_logger import DDLogFormatter, DDLogger
 

scfw/loggers/dd_logger.py

@@ -11,7 +11,7 @@
 import dotenv
 
 import scfw
-from scfw.configure import DD_ENV, DD_LOG_LEVEL_VAR, DD_SERVICE, DD_SOURCE
+from scfw.constants import DD_ENV, DD_LOG_LEVEL_VAR, DD_SERVICE, DD_SOURCE
 from scfw.ecosystem import ECOSYSTEM
 from scfw.logger import FirewallAction, FirewallLogger
 from scfw.package import Package