ci: add pull request validation
This commit is contained in:
Jason Morcos
Executable
+126
@@ -0,0 +1,126 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Verify that SmartThings-Local wheel and sdist contents are intentional."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import subprocess
|
||||
import tarfile
|
||||
import zipfile
|
||||
from pathlib import Path, PurePosixPath
|
||||
|
||||
|
||||
class DistributionError(RuntimeError):
|
||||
"""An artifact contains a missing, unexpected, or unsafe member."""
|
||||
|
||||
|
||||
def _tracked_files() -> set[str]:
|
||||
proc = subprocess.run(
|
||||
["git", "ls-files", "-z", "--", "smartthings_local", "tests"],
|
||||
capture_output=True,
|
||||
check=True,
|
||||
)
|
||||
return {value.decode("utf-8") for value in proc.stdout.split(b"\0") if value}
|
||||
|
||||
|
||||
def _safe_member(name: str) -> bool:
|
||||
path = PurePosixPath(name)
|
||||
return bool(name) and not path.is_absolute() and ".." not in path.parts
|
||||
|
||||
|
||||
def _expected_package_files() -> set[str]:
|
||||
tracked = {
|
||||
path for path in _tracked_files() if path.startswith("smartthings_local/")
|
||||
}
|
||||
tracked.add("smartthings_local/_version.py")
|
||||
return tracked
|
||||
|
||||
|
||||
def check_wheel(path: Path) -> None:
|
||||
with zipfile.ZipFile(path) as archive:
|
||||
names = set(archive.namelist())
|
||||
if not names or any(not _safe_member(name) for name in names):
|
||||
raise DistributionError("wheel has an unsafe member")
|
||||
|
||||
package_files = {name for name in names if name.startswith("smartthings_local/")}
|
||||
if package_files != _expected_package_files():
|
||||
raise DistributionError(
|
||||
"wheel package contents differ from the tracked package"
|
||||
)
|
||||
|
||||
metadata = names - package_files
|
||||
roots = {name.split("/", 1)[0] for name in metadata}
|
||||
if len(roots) != 1:
|
||||
raise DistributionError("wheel must contain one dist-info directory")
|
||||
dist_info = roots.pop()
|
||||
if not dist_info.endswith(".dist-info"):
|
||||
raise DistributionError("wheel metadata directory is invalid")
|
||||
expected_metadata = {
|
||||
f"{dist_info}/METADATA",
|
||||
f"{dist_info}/WHEEL",
|
||||
f"{dist_info}/licenses/LICENSE",
|
||||
f"{dist_info}/RECORD",
|
||||
}
|
||||
if metadata != expected_metadata:
|
||||
raise DistributionError("wheel metadata contents are unexpected")
|
||||
|
||||
|
||||
def check_sdist(path: Path) -> None:
|
||||
with tarfile.open(path, mode="r:gz") as archive:
|
||||
members = archive.getmembers()
|
||||
if not members or any(
|
||||
not member.isfile() or member.issym() or member.islnk() for member in members
|
||||
):
|
||||
raise DistributionError("sdist must contain regular files only")
|
||||
names = {member.name for member in members}
|
||||
if any(not _safe_member(name) for name in names):
|
||||
raise DistributionError("sdist has an unsafe member")
|
||||
|
||||
roots = {name.split("/", 1)[0] for name in names}
|
||||
if len(roots) != 1:
|
||||
raise DistributionError("sdist must contain one top-level directory")
|
||||
root = roots.pop()
|
||||
relative = {name[len(root) + 1 :] for name in names if name.startswith(f"{root}/")}
|
||||
expected = _tracked_files() | {
|
||||
".gitignore",
|
||||
".hgignore",
|
||||
"LICENSE",
|
||||
"PKG-INFO",
|
||||
"README.md",
|
||||
"pyproject.toml",
|
||||
"smartthings_local/_version.py",
|
||||
}
|
||||
if relative != expected:
|
||||
raise DistributionError("sdist contents differ from the intended source set")
|
||||
|
||||
|
||||
def check_directory(directory: Path) -> None:
|
||||
wheels = sorted(directory.glob("*.whl"))
|
||||
sdists = sorted(directory.glob("*.tar.gz"))
|
||||
if len(wheels) != 1 or len(sdists) != 1:
|
||||
raise DistributionError("expected exactly one wheel and one sdist")
|
||||
check_wheel(wheels[0])
|
||||
check_sdist(sdists[0])
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("directory", type=Path)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
check_directory(args.directory)
|
||||
except (
|
||||
DistributionError,
|
||||
OSError,
|
||||
subprocess.SubprocessError,
|
||||
tarfile.TarError,
|
||||
zipfile.BadZipFile,
|
||||
):
|
||||
print("distribution check failed")
|
||||
return 1
|
||||
print("distribution contents verified")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Executable
+229
@@ -0,0 +1,229 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check changed public files for common private-data and secret shapes.
|
||||
|
||||
Findings contain only path, line, and rule ID. Matched content is never
|
||||
printed because it may itself be sensitive.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ipaddress
|
||||
import re
|
||||
import subprocess
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
MAX_TEXT_BYTES = 2 * 1024 * 1024
|
||||
DOCUMENTATION_IPV4 = tuple(
|
||||
ipaddress.ip_network(value)
|
||||
for value in ("192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24")
|
||||
)
|
||||
DOCUMENTATION_IPV6 = ipaddress.ip_network("2001:db8::/32")
|
||||
SAFE_UUIDS = {
|
||||
"00000000-0000-0000-0000-000000000000",
|
||||
"11111111-2222-3333-4444-555555555555",
|
||||
}
|
||||
FORBIDDEN_SUFFIXES = {
|
||||
".7z",
|
||||
".apk",
|
||||
".cap",
|
||||
".db",
|
||||
".der",
|
||||
".gz",
|
||||
".jks",
|
||||
".key",
|
||||
".p12",
|
||||
".pcap",
|
||||
".pcapng",
|
||||
".pfx",
|
||||
".sqlite",
|
||||
".sqlite3",
|
||||
".tar",
|
||||
".tgz",
|
||||
".zip",
|
||||
}
|
||||
|
||||
PATTERNS = (
|
||||
(
|
||||
"PEM_PRIVATE_KEY",
|
||||
re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----"),
|
||||
),
|
||||
(
|
||||
"EMAIL_ADDRESS",
|
||||
re.compile(r"[A-Za-z0-9._%+-]+@(?:[A-Za-z0-9-]+\.)+[A-Za-z]{2,}"),
|
||||
),
|
||||
(
|
||||
"MAC_ADDRESS",
|
||||
re.compile(r"(?i)(?<![0-9a-f])(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}(?![0-9a-f])"),
|
||||
),
|
||||
(
|
||||
"PRIVATE_DNS",
|
||||
re.compile(r"(?i)\b(?:[a-z0-9-]+\.)+(?:corp|home|internal|lan|local)\b"),
|
||||
),
|
||||
("HOME_PATH", re.compile(r"(?<![A-Za-z0-9._-])/(?:Users|home)/[^\s'\"`]+")),
|
||||
(
|
||||
"CREDENTIAL_URL",
|
||||
re.compile(
|
||||
r"(?i)\bhttps?://(?:[^\s/@:]+:[^\s/@]+@|[^\s?#]+[?&](?:access_token|api_key|password|refresh_token|token)=)"
|
||||
),
|
||||
),
|
||||
(
|
||||
"SECRET_ASSIGNMENT",
|
||||
re.compile(
|
||||
r"(?i)\b(?:access[_-]?token|api[_-]?key|bearer|owner[_-]?psk|password|passwd|private[_-]?key|psk|refresh[_-]?token|secret)\b\s*(?::|=)\s*(?:b|br|f|r|rb)?['\"][^'\"]+['\"]"
|
||||
),
|
||||
),
|
||||
(
|
||||
"SERIAL_ASSIGNMENT",
|
||||
re.compile(
|
||||
r"(?i)\b(?:device[_-]?)?serial(?:number|num)?\b\s*(?::|=)\s*['\"][^'\"]+['\"]"
|
||||
),
|
||||
),
|
||||
(
|
||||
"REAL_TIMESTAMP",
|
||||
re.compile(
|
||||
r"\b20[0-9]{2}-[01][0-9]-[0-3][0-9][T ][0-2][0-9]:[0-5][0-9](?::[0-6][0-9](?:\.[0-9]+)?)?(?:Z|[+-][0-2][0-9]:?[0-5][0-9])?\b"
|
||||
),
|
||||
),
|
||||
(
|
||||
"QR_PAYLOAD",
|
||||
re.compile(r"(?i)\b(?:qr[_-]?payload|setup[_-]?payload)\b\s*(?::|=)"),
|
||||
),
|
||||
)
|
||||
UUID_RE = re.compile(
|
||||
r"(?i)(?<![0-9a-f])[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}(?![0-9a-f])"
|
||||
)
|
||||
IPV4_RE = re.compile(
|
||||
r"(?<![0-9])(?:25[0-5]|2[0-4][0-9]|1?[0-9]{1,2})(?:\.(?:25[0-5]|2[0-4][0-9]|1?[0-9]{1,2})){3}(?![0-9])"
|
||||
)
|
||||
IPV6_RE = re.compile(
|
||||
r"(?i)(?<![0-9a-f:])(?:\[)?(?:[0-9a-f]{0,4}:){2,7}[0-9a-f]{0,4}(?:%[A-Za-z0-9_.-]+)?(?:\])?(?![0-9a-f:])"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, order=True)
|
||||
class Finding:
|
||||
path: str
|
||||
line: int
|
||||
rule_id: str
|
||||
|
||||
def render(self) -> str:
|
||||
return f"{self.path}:{self.line}:{self.rule_id}"
|
||||
|
||||
|
||||
def _safe_ipv4(value: str) -> bool:
|
||||
address = ipaddress.ip_address(value)
|
||||
return (
|
||||
address.is_loopback
|
||||
or address.is_unspecified
|
||||
or any(address in network for network in DOCUMENTATION_IPV4)
|
||||
)
|
||||
|
||||
|
||||
def _safe_ipv6(value: str) -> bool:
|
||||
address = ipaddress.ip_address(value.strip("[]").split("%", 1)[0])
|
||||
return (
|
||||
address.is_loopback or address.is_unspecified or address in DOCUMENTATION_IPV6
|
||||
)
|
||||
|
||||
|
||||
def scan_text(path: str, text: str) -> list[Finding]:
|
||||
findings: set[Finding] = set()
|
||||
for line_number, line in enumerate(text.splitlines(), start=1):
|
||||
for rule_id, pattern in PATTERNS:
|
||||
if pattern.search(line):
|
||||
findings.add(Finding(path, line_number, rule_id))
|
||||
for match in UUID_RE.finditer(line):
|
||||
if match.group(0).lower() not in SAFE_UUIDS:
|
||||
findings.add(Finding(path, line_number, "UUID"))
|
||||
for match in IPV4_RE.finditer(line):
|
||||
if not _safe_ipv4(match.group(0)):
|
||||
findings.add(Finding(path, line_number, "NON_DOCUMENTATION_IPV4"))
|
||||
for match in IPV6_RE.finditer(line):
|
||||
try:
|
||||
safe = _safe_ipv6(match.group(0))
|
||||
except ValueError:
|
||||
continue
|
||||
if not safe:
|
||||
findings.add(Finding(path, line_number, "NON_DOCUMENTATION_IPV6"))
|
||||
return sorted(findings)
|
||||
|
||||
|
||||
def scan_file(path: Path, display_path: str) -> list[Finding]:
|
||||
if path.is_symlink():
|
||||
return [Finding(display_path, 0, "SYMLINK")]
|
||||
if path.suffix.casefold() in FORBIDDEN_SUFFIXES:
|
||||
return [Finding(display_path, 0, "FORBIDDEN_FILE_TYPE")]
|
||||
data = path.read_bytes()
|
||||
if len(data) > MAX_TEXT_BYTES:
|
||||
return [Finding(display_path, 0, "FILE_TOO_LARGE")]
|
||||
if b"\x00" in data:
|
||||
return [Finding(display_path, 0, "BINARY_CONTENT")]
|
||||
try:
|
||||
text = data.decode("utf-8", errors="strict")
|
||||
except UnicodeDecodeError:
|
||||
return [Finding(display_path, 0, "NON_UTF8_CONTENT")]
|
||||
return scan_text(display_path, text)
|
||||
|
||||
|
||||
def _changed_paths(base: str) -> list[str]:
|
||||
commands = (
|
||||
[
|
||||
"git",
|
||||
"diff",
|
||||
"--name-only",
|
||||
"--diff-filter=ACMR",
|
||||
"-z",
|
||||
f"{base}..HEAD",
|
||||
"--",
|
||||
],
|
||||
["git", "diff", "--cached", "--name-only", "--diff-filter=ACMR", "-z", "--"],
|
||||
["git", "diff", "--name-only", "--diff-filter=ACMR", "-z", "--"],
|
||||
)
|
||||
changed = [
|
||||
subprocess.run(command, capture_output=True, check=True) for command in commands
|
||||
]
|
||||
untracked = subprocess.run(
|
||||
["git", "ls-files", "--others", "--exclude-standard", "-z"],
|
||||
capture_output=True,
|
||||
check=True,
|
||||
)
|
||||
return sorted(
|
||||
{
|
||||
value.decode("utf-8")
|
||||
for output in (*(result.stdout for result in changed), untracked.stdout)
|
||||
for value in output.split(b"\0")
|
||||
if value
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def check_paths(paths: Iterable[str]) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for value in paths:
|
||||
findings.extend(scan_file(Path(value), value))
|
||||
return sorted(set(findings))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("paths", nargs="*")
|
||||
parser.add_argument("--changed-since")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
paths = _changed_paths(args.changed_since) if args.changed_since else args.paths
|
||||
if not paths:
|
||||
raise ValueError("no paths selected")
|
||||
findings = check_paths(paths)
|
||||
except (OSError, UnicodeDecodeError, subprocess.SubprocessError, ValueError):
|
||||
print("share-safety scan failed closed:SCAN_ERROR")
|
||||
return 2
|
||||
for finding in findings:
|
||||
print(finding.render())
|
||||
return 1 if findings else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user