Add audit (#17951)

Changelog: Omit
Docs: Omit

All merit goes to @AbrilRBS! I had to squash her commits because of the
CLA

supersedes
#17950 (review)
that had a problem with the CLA

Author: memsharded

conan/api/conan_api.py

@@ -2,6 +2,7 @@
 import sys
 
 from conan.api.output import init_colorama
+from conan.api.subapi.audit import AuditAPI
 from conan.api.subapi.cache import CacheAPI
 from conan.api.subapi.command import CommandAPI
 from conan.api.subapi.local import LocalAPI
@@ -69,6 +70,7 @@ def __init__(self, cache_folder=None):
         self.cache = CacheAPI(self)
         self.lockfile = LockfileAPI(self)
         self.local = LocalAPI(self)
+        self.audit = AuditAPI(self)
 
         _check_conan_version(self)
 

conan/api/subapi/audit.py

@@ -0,0 +1,159 @@
+import binascii
+import json
+import os
+import base64
+
+from conan.internal.api.audit.providers import ConanCenterProvider, PrivateProvider
+from conan.errors import ConanException
+from conan.internal.api.remotes.encrypt import encode, decode
+from conan.internal.model.recipe_ref import RecipeReference
+from conans.util.files import save, load
+
+CONAN_CENTER_AUDIT_PROVIDER_NAME = "conancenter"
+CYPHER_KEY = "private"
+
+class AuditAPI:
+    """
+    This class provides the functionality to scan references for vulnerabilities.
+    """
+
+    def __init__(self, conan_api):
+        self.conan_api = conan_api
+        self._home_folder = conan_api.home_folder
+        self._providers_path = os.path.join(self._home_folder, "audit_providers.json")
+        self._provider_cls = {
+            "conan-center-proxy": ConanCenterProvider,
+            "private": PrivateProvider,
+        }
+
+    def scan(self, deps_graph, provider):
+        """
+        Scan a given recipe for vulnerabilities in its dependencies.
+        """
+        refs = sorted(set(RecipeReference.loads(f"{node.ref.name}/{node.ref.version}")
+                        for node in deps_graph.nodes[1:]), key= lambda ref: ref.name)
+        return provider.get_cves(refs)
+
+    def list(self, references, provider):
+        """
+        List the vulnerabilities of the given reference.
+        """
+        refs = [RecipeReference.loads(ref) for ref in references]
+        for ref in refs:
+            ref.validate_ref()
+        return provider.get_cves(refs)
+
+    def get_provider(self, provider_name):
+        """
+        Get the provider by name.
+        """
+        # TODO: More work remains to be done here, hardcoded for now for testing
+        providers = _load_providers(self._providers_path)
+        if provider_name not in providers:
+            add_arguments = (
+                "--url=https://audit.conan.io/ --type=conan-center-proxy"
+                if provider_name == CONAN_CENTER_AUDIT_PROVIDER_NAME
+                else "--url=<url> --type=<type>"
+            )
+
+            register_message = (
+                f"If you don't have a valid token, register at: https://audit.conan.io/register."
+                if provider_name == CONAN_CENTER_AUDIT_PROVIDER_NAME
+                else ""
+            )
+
+            raise ConanException(
+                f"Provider '{provider_name}' not found. Please specify a valid provider name or add it using: "
+                f"'conan audit provider add {provider_name} {add_arguments} --token=<token>'\n"
+                f"{register_message}"
+            )
+
+        provider_data = providers[provider_name]
+        safe_provider_name = provider_name.replace("-", "_")
+        env_token = os.getenv(f"CONAN_AUDIT_PROVIDER_TOKEN_{safe_provider_name.upper()}")
+
+        if env_token:
+            # Always override the token with the environment variable
+            provider_data["token"] = env_token
+        elif "token" in provider_data:
+            try:
+                provider_data["token"] = decode(base64.standard_b64decode(provider_data["token"]).decode(), CYPHER_KEY)
+            except binascii.Error as e:
+                raise ConanException(f"Invalid token format for provider '{provider_name}'. The token might be corrupt.")
+
+        provider_cls = self._provider_cls.get(provider_data["type"])
+
+        return provider_cls(self.conan_api, provider_name, provider_data)
+
+    def list_providers(self):
+        """
+        Get all available providers.
+        """
+        providers = _load_providers(self._providers_path)
+        result = []
+        for name, provider_data in providers.items():
+            provider_cls = self._provider_cls.get(provider_data["type"])
+            result.append(provider_cls(self.conan_api, name, provider_data))
+        return result
+
+    def add_provider(self, name, url, provider_type):
+        """
+        Add a provider.
+        """
+        providers = _load_providers(self._providers_path)
+        if name in providers:
+            raise ConanException(f"Provider '{name}' already exists")
+
+        if provider_type not in self._provider_cls:
+            raise ConanException(f"Provider type '{provider_type}' not found")
+
+        providers[name] = {
+            "name": name,
+            "url": url,
+            "type": provider_type
+        }
+
+        _save_providers(self._providers_path, providers)
+
+    def remove_provider(self, provider_name):
+        """
+        Remove a provider.
+        """
+        providers = _load_providers(self._providers_path)
+        if provider_name not in providers:
+            raise ConanException(f"Provider '{provider_name}' not found")
+
+        del providers[provider_name]
+
+        _save_providers(self._providers_path, providers)
+
+    def auth_provider(self, provider, token):
+        """
+        Authenticate a provider.
+        """
+        if not provider:
+            raise ConanException("Provider not found")
+
+        providers = _load_providers(self._providers_path)
+
+        assert provider.name in providers
+        providers[provider.name]["token"] = base64.standard_b64encode(encode(token, CYPHER_KEY).encode()).decode()
+        setattr(provider, "token", token)
+        _save_providers(self._providers_path, providers)
+
+
+def _load_providers(providers_path):
+    if not os.path.exists(providers_path):
+        default_providers = {
+            CONAN_CENTER_AUDIT_PROVIDER_NAME: {
+                "url": "https://audit.conan.io/",
+                "type": "conan-center-proxy"
+            }
+        }
+        save(providers_path, json.dumps(default_providers, indent=4))
+
+    return json.loads(load(providers_path))
+
+
+def _save_providers(providers_path, providers):
+    save(providers_path, json.dumps(providers, indent=4))

conan/cli/commands/audit.py

@@ -0,0 +1,195 @@
+import json
+import os
+
+from conan.api.conan_api import ConanAPI
+from conan.api.input import UserInput
+from conan.api.model import MultiPackagesList
+from conan.api.output import cli_out_write, ConanOutput
+from conan.api.subapi.audit import CONAN_CENTER_AUDIT_PROVIDER_NAME
+from conan.cli import make_abs_path
+from conan.cli.args import common_graph_args, validate_common_graph_args
+from conan.cli.command import conan_command, conan_subcommand
+from conan.cli.formatters.audit.vulnerabilities import text_vuln_formatter, json_vuln_formatter, \
+    html_vuln_formatter
+from conan.cli.printers import print_profiles
+from conan.cli.printers.graph import print_graph_basic
+from conan.errors import ConanException
+
+
+def _add_provider_arg(subparser):
+    subparser.add_argument("-p", "--provider", help="Provider to use for scanning")
+
+
+@conan_subcommand(formatters={"text": text_vuln_formatter,
+                              "json": json_vuln_formatter,
+                              "html": html_vuln_formatter})
+def audit_scan(conan_api: ConanAPI, parser, subparser, *args):
+    """
+    Scan a given recipe for vulnerabilities in its dependencies.
+    """
+    common_graph_args(subparser)
+    # Needed for the validation of args, but this should usually be left as False here
+    # TODO: Do we then want to hide it in the --help?
+    subparser.add_argument("--build-require", action='store_true', default=False,
+                           help='Whether the provided reference is a build-require')
+
+    _add_provider_arg(subparser)
+    args = parser.parse_args(*args)
+
+    # This comes from install command
+
+    validate_common_graph_args(args)
+    # basic paths
+    cwd = os.getcwd()
+    path = conan_api.local.get_conanfile_path(args.path, cwd, py=None) if args.path else None
+
+    # Basic collaborators: remotes, lockfile, profiles
+    remotes = conan_api.remotes.list(args.remote) if not args.no_remote else []
+    overrides = eval(args.lockfile_overrides) if args.lockfile_overrides else None
+    lockfile = conan_api.lockfile.get_lockfile(lockfile=args.lockfile, conanfile_path=path, cwd=cwd,
+                                               partial=args.lockfile_partial, overrides=overrides)
+    profile_host, profile_build = conan_api.profiles.get_profiles_from_args(args)
+    print_profiles(profile_host, profile_build)
+
+    # Graph computation (without installation of binaries)
+    gapi = conan_api.graph
+    if path:
+        deps_graph = gapi.load_graph_consumer(path, args.name, args.version, args.user, args.channel,
+                                              profile_host, profile_build, lockfile, remotes,
+                                              # TO DISCUSS: defaulting to False the is_build_require
+                                              args.update, is_build_require=args.build_require)
+    else:
+        deps_graph = gapi.load_graph_requires(args.requires, args.tool_requires, profile_host,
+                                              profile_build, lockfile, remotes, args.update)
+    print_graph_basic(deps_graph)
+    deps_graph.report_graph_error()
+
+    if deps_graph.error:
+        return {"error": deps_graph.error}
+
+    provider = conan_api.audit.get_provider(args.provider or CONAN_CENTER_AUDIT_PROVIDER_NAME)
+
+    return conan_api.audit.scan(deps_graph, provider)
+
+
+@conan_subcommand(formatters={"text": text_vuln_formatter,
+                              "json": json_vuln_formatter,
+                              "html": html_vuln_formatter})
+def audit_list(conan_api: ConanAPI, parser, subparser, *args):
+    """
+    List the vulnerabilities of the given reference.
+    """
+    subparser.add_argument("reference", help="Reference to list vulnerabilities for", nargs="?")
+    subparser.add_argument("-l", "--list", help="pkglist file to list vulnerabilities for", default=None)
+    subparser.add_argument("-r", "--remote", help="Remote to use for listing", default=None)
+    _add_provider_arg(subparser)
+    args = parser.parse_args(*args)
+
+    if not args.reference and not args.list:
+        raise ConanException("Please specify a reference or a pkglist file")
+
+    if args.reference and args.list:
+        raise ConanException("Please specify a reference or a pkglist file, not both")
+
+    provider = conan_api.audit.get_provider(args.provider or CONAN_CENTER_AUDIT_PROVIDER_NAME)
+
+    references = []
+    if args.list:
+        listfile = make_abs_path(args.list)
+        multi_package_list = MultiPackagesList.load(listfile)
+        cache_name = "Local Cache" if not args.remote else args.remote
+        package_list = multi_package_list[cache_name]
+        refs_to_list = package_list.serialize()
+        references = list(refs_to_list.keys())
+        if not references:  # the package list might contain only refs, no revs
+            ConanOutput().warning("Nothing to list, package list do not contain recipe revisions")
+    else:
+        references = [args.reference]
+    return conan_api.audit.list(references, provider)
+
+def text_provider_formatter(providers_action):
+    providers = providers_action[0]
+    action = providers_action[1]
+
+    if action == "remove":
+        cli_out_write("Provider removed successfully.")
+    elif action == "add":
+        cli_out_write("Provider added successfully.")
+    elif action == "auth":
+        cli_out_write("Provider authentication added.")
+    elif action == "list":
+        if not providers:
+            cli_out_write("No providers found.")
+        else:
+            for provider in providers:
+                if provider:
+                    cli_out_write(f"{provider.name} (type: {provider.type}) - {provider.url}")
+
+def json_provider_formatter(providers_action):
+    ret = []
+    for provider in providers_action[0]:
+        if provider:
+            ret.append({"name": provider.name, "url": provider.url, "type": provider.type})
+    cli_out_write(json.dumps(ret, indent=4))
+
+
+@conan_subcommand(formatters={"text": text_provider_formatter, "json": json_provider_formatter})
+def audit_provider(conan_api, parser, subparser, *args):
+    """
+    Manage security providers for the 'conan audit' command.
+    """
+
+    subparser.add_argument("action", choices=["add", "list", "auth", "remove"], help="Action to perform from 'add', 'list' , 'remove' or 'auth'")
+    subparser.add_argument("name", help="Provider name", nargs="?")
+
+    subparser.add_argument("--url", help="Provider URL")
+    subparser.add_argument("--type", help="Provider type", choices=["conan-center-proxy", "private"])
+    subparser.add_argument("--token", help="Provider token")
+    args = parser.parse_args(*args)
+
+    if args.action == "add":
+        if not args.name or not args.url or not args.type:
+            raise ConanException("Name, URL and type are required to add a provider")
+        if " " in args.name:
+            raise ConanException("Name cannot contain spaces")
+        conan_api.audit.add_provider(args.name, args.url, args.type)
+
+        if not args.token:
+            user_input = UserInput(conan_api.config.get("core:non_interactive"))
+            ConanOutput().write(f"Please enter a token for {args.name} the provider: ")
+            token = user_input.get_password()
+        else:
+            token = args.token
+
+        provider = conan_api.audit.get_provider(args.name)
+        if token:
+            conan_api.audit.auth_provider(provider, token)
+
+        return [provider], args.action
+    elif args.action == "remove":
+        if not args.name:
+            raise ConanException("Name required to remove a provider")
+        conan_api.audit.remove_provider(args.name)
+        return [], args.action
+    elif args.action == "list":
+        providers = conan_api.audit.list_providers()
+        return providers, args.action
+    elif args.action == "auth":
+        if not args.name:
+            raise ConanException("Name is required to authenticate on a provider")
+        if not args.token:
+            user_input = UserInput(conan_api.config.get("core:non_interactive"))
+            ConanOutput().write(f"Please enter a token for {args.name} the provider: ")
+            token = user_input.get_password()
+        else:
+            token = args.token
+
+        provider = conan_api.audit.get_provider(args.name)
+        conan_api.audit.auth_provider(provider, token)
+        return [provider], args.action
+
+@conan_command(group="Security")
+def audit(conan_api, parser, *args):
+    """
+    Find vulnerabilities in your dependencies.
+    """