Source code
Revision control
Copy as Markdown
Other Tools
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import json
import os
import subprocess
from mozbuild.nodeutil import find_node_executable
from mozlint import result
from mozlint.pathutils import expand_exclusions
ERROR_SEVERITIES = ("critical", "high")
SHOWN_PATHS = 3
NO_NODE_MESSAGE = """
Could not find a node executable. Run `mach bootstrap` and try again.
""".strip()
NO_PNPM_MESSAGE = """
Could not find or bootstrap pnpm. Check the output above and try again.
""".strip()
REGISTRY_MESSAGE = """
This usually means pnpm could not reach the npm registry to download the
advisory database, which evolves independently of this repository. Please file
""".strip()
def to_str_paths(finding, verbose):
paths = finding.get("paths") or []
if not paths:
return ""
shown = paths if verbose else paths[:SHOWN_PATHS]
lines = [f"\n {path}" for path in shown]
remaining = len(paths) - len(shown)
if remaining:
lines.append(f"\n and {remaining} more, re-run with -v to see them all")
return "\nDependency paths:" + "".join(lines)
def build_message(advisory, verbose):
module = advisory["module_name"]
message = f"Depends on a vulnerable version of {module}."
message += f"\n\nAdvisory:\n{advisory['title']}"
message += f"\nPackage: {module}"
identifier = advisory.get("github_advisory_id") or advisory.get("id")
if identifier:
message += f"\nID: {identifier}"
message += f"\nSeverity: {advisory['severity']}"
cwe = advisory.get("cwe")
if cwe:
message += f"\nCWE: {cwe}"
url = advisory.get("url")
if url:
message += f"\nURL: {url}"
findings = advisory.get("findings") or []
installed = sorted({
finding["version"] for finding in findings if "version" in finding
})
if installed:
message += f"\n\nInstalled versions: {', '.join(installed)}"
for key, label in (
("vulnerable_versions", "Vulnerable versions"),
("patched_versions", "Patched versions"),
):
if advisory.get(key):
message += f"\n{label}: {advisory[key]}"
for finding in findings:
message += to_str_paths(finding, verbose)
return message
def severity_level(severity):
return "error" if severity in ERROR_SEVERITIES else "warning"
def build_issue(config, path, message, level):
return result.from_config(
config,
**{
"path": path,
"message": message,
"lineno": -1,
"column": -1,
"level": level,
},
)
def is_excluded(message, exclusions):
return any(exclusion in message for exclusion in exclusions)
def locate_pnpm():
from mozbuild.bootstrap import bootstrap_toolchain
return bootstrap_toolchain("pnpm/bin/pnpm.cjs")
def audit_failed(args, completed, reason):
message = [
f"pnpm audit {reason} (exit code {completed.returncode}) while running:",
" " + " ".join(args),
]
for name, stream in (("stdout", completed.stdout), ("stderr", completed.stderr)):
if stream.strip():
message.append(f"\n{name}:")
message.append(stream.rstrip())
message.append(f"\n{REGISTRY_MESSAGE}")
return RuntimeError("\n".join(message))
def run_audit(node, pnpm, directory):
args = [node, pnpm, "audit", "--json", "--dir", directory]
completed = subprocess.run(
args,
capture_output=True,
text=True,
check=False,
)
try:
report = json.loads(completed.stdout)
except json.JSONDecodeError:
raise audit_failed(args, completed, "did not return JSON")
if not isinstance(report, dict):
raise audit_failed(args, completed, "did not return a report")
error = report.get("error")
if error:
detail = error.get("message", error) if isinstance(error, dict) else error
raise audit_failed(args, completed, f"reported an error, {detail}")
if not isinstance(report.get("advisories"), dict):
raise audit_failed(args, completed, "returned no advisories section")
return report
def lint(paths, config, log, **lintargs):
node, _ = find_node_executable()
if not node:
raise RuntimeError(NO_NODE_MESSAGE)
pnpm = locate_pnpm()
if not pnpm:
raise RuntimeError(NO_PNPM_MESSAGE)
verbose = lintargs.get("show_verbose", False)
exclusions = config.get("exclude-error", [])
results = []
for path in expand_exclusions(paths, config, lintargs["root"]):
report = run_audit(node, pnpm, os.path.dirname(path))
for advisory in report["advisories"].values():
message = build_message(advisory, verbose)
if is_excluded(message, exclusions):
continue
level = severity_level(advisory["severity"])
results.append(build_issue(config, path, message, level))
return results
def setup(root, log, **lintargs):
node, _ = find_node_executable()
if not node:
log.error(NO_NODE_MESSAGE)
return 1
if not locate_pnpm():
log.error(NO_PNPM_MESSAGE)
return 1
return 0