music-assistant-server

1#!/usr/bin/env python3
2"""
3Check PyPI package metadata for security and supply chain concerns.
4
5This script checks new or updated Python dependencies for suspicious indicators
6that might suggest supply chain attacks or unmaintained packages.
7"""
8
9# ruff: noqa: T201, S310, RUF001, PLR0915
10import json
11import re
12import sys
13import urllib.request
14from datetime import datetime
15from typing import Any
16
17# OSI-approved and common compatible licenses
18COMPATIBLE_LICENSES = {
19    "MIT",
20    "Apache-2.0",
21    "Apache Software License",
22    "BSD",
23    "BSD-3-Clause",
24    "BSD-2-Clause",
25    "ISC",
26    "Python Software Foundation License",
27    "PSF",
28    "LGPL",
29    "MPL-2.0",
30    "Unlicense",
31    "CC0",
32}
33
34# Common packages to check for typosquatting (popular Python packages)
35POPULAR_PACKAGES = {
36    "requests",
37    "urllib3",
38    "setuptools",
39    "certifi",
40    "pip",
41    "numpy",
42    "pandas",
43    "boto3",
44    "botocore",
45    "awscli",
46    "django",
47    "flask",
48    "sqlalchemy",
49    "pytest",
50    "pydantic",
51    "aiohttp",
52    "fastapi",
53}
54
55
56def check_typosquatting(package_name: str) -> str | None:
57    """
58    Check if package name might be typosquatting a popular package.
59
60    :param package_name: The package name to check.
61    """
62    package_lower = package_name.lower().replace("-", "").replace("_", "")
63
64    for popular in POPULAR_PACKAGES:
65        popular_normalized = popular.lower().replace("-", "").replace("_", "")
66
67        # Check for common typosquatting techniques
68        if package_lower == popular_normalized:
69            continue  # Exact match is fine
70
71        # Check edit distance (1-2 character changes)
72        if len(package_lower) == len(popular_normalized):
73            differences = sum(
74                c1 != c2 for c1, c2 in zip(package_lower, popular_normalized, strict=True)
75            )
76            if differences == 1:
77                return f"Suspicious: Very similar to popular package '{popular}'"
78
79        # Check for common substitutions
80        substitutions = [
81            ("0", "o"),
82            ("1", "l"),
83            ("1", "i"),
84        ]
85        for old, new in substitutions:
86            if old in package_lower:
87                test_name = package_lower.replace(old, new)
88                if test_name == popular_normalized:
89                    return f"Suspicious: Character substitution of popular package '{popular}'"
90
91    return None
92
93
94def check_license_compatibility(license_str: str) -> tuple[bool, str]:
95    """
96    Check if license is compatible with the project.
97
98    :param license_str: The license string from PyPI.
99    """
100    if not license_str or license_str == "Unknown":
101        return False, "No license information"
102
103    license_upper = license_str.upper()
104
105    # Check against compatible licenses
106    for compatible in COMPATIBLE_LICENSES:
107        if compatible.upper() in license_upper:
108            return True, f"Compatible ({license_str})"
109
110    # Check for problematic licenses
111    problematic = ["GPL", "AGPL", "SSPL"]
112    for problem in problematic:
113        if problem in license_upper and "LGPL" not in license_upper:
114            return False, f"Incompatible copyleft license ({license_str})"
115
116    # Unknown license
117    return False, f"Unknown/unverified license ({license_str})"
118
119
120def parse_requirement(line: str) -> str | None:
121    """
122    Extract package name from a requirement line.
123
124    :param line: A line from requirements.txt (e.g., "package==1.0.0" or "package>=1.0")
125    """
126    line = line.strip()
127    if not line or line.startswith("#"):
128        return None
129
130    # Handle various requirement formats
131    # package==1.0.0, package>=1.0, package[extra]>=1.0, etc.
132    match = re.match(r"^([a-zA-Z0-9_-]+)", line)
133    if match:
134        return match.group(1).lower()
135    return None
136
137
138def get_pypi_metadata(package_name: str) -> dict[str, Any] | None:
139    """
140    Fetch package metadata from PyPI JSON API.
141
142    :param package_name: The name of the package to check.
143    """
144    url = f"https://pypi.org/pypi/{package_name}/json"
145
146    try:
147        with urllib.request.urlopen(url, timeout=10) as response:
148            return json.loads(response.read())
149    except urllib.error.HTTPError as err:
150        if err.code == 404:
151            print(f"❌ Package '{package_name}' not found on PyPI")
152        else:
153            print(f"⚠️  Error fetching metadata for '{package_name}': {err}")
154        return None
155    except Exception as err:
156        print(f"⚠️  Error fetching metadata for '{package_name}': {err}")
157        return None
158
159
160def check_package(package_name: str) -> dict[str, Any]:
161    """
162    Check a single package for security concerns.
163
164    :param package_name: The name of the package to check.
165    """
166    data = get_pypi_metadata(package_name)
167
168    if not data:
169        return {
170            "name": package_name,
171            "error": "Could not fetch package metadata",
172            "risk_level": "unknown",
173            "warnings": [],
174        }
175
176    info = data.get("info", {})
177    releases = data.get("releases", {})
178
179    # Get package age
180    upload_times = []
181    for release_files in releases.values():
182        if release_files:
183            for file_info in release_files:
184                if "upload_time" in file_info:
185                    try:
186                        upload_time_str = file_info["upload_time"]
187                        # Handle both formats: with 'Z' suffix or with timezone
188                        if upload_time_str.endswith("Z"):
189                            upload_time_str = upload_time_str[:-1] + "+00:00"
190                        upload_time = datetime.fromisoformat(upload_time_str)
191                        upload_times.append(upload_time)
192                    except ValueError, AttributeError:
193                        continue
194
195    first_upload = min(upload_times) if upload_times else None
196    age_days = (datetime.now(first_upload.tzinfo) - first_upload).days if first_upload else 0
197
198    # Extract metadata
199    project_urls = info.get("project_urls") or {}
200    homepage = info.get("home_page") or project_urls.get("Homepage")
201    source = project_urls.get("Source") or project_urls.get("Repository")
202
203    # Run automated security checks
204    typosquat_check = check_typosquatting(package_name)
205    license_compatible, license_status = check_license_compatibility(info.get("license", "Unknown"))
206
207    checks = {
208        "name": package_name,
209        "version": info.get("version", "unknown"),
210        "age_days": age_days,
211        "total_releases": len(releases),
212        "has_homepage": bool(homepage),
213        "has_source": bool(source),
214        "author": info.get("author") or info.get("maintainer") or "Unknown",
215        "license": info.get("license") or "Unknown",
216        "summary": info.get("summary", "No description"),
217        "warnings": [],
218        "info_items": [],
219        "risk_level": "low",
220        "automated_checks": {
221            "trusted_source": bool(source),
222            "typosquatting": typosquat_check is None,
223            "license_compatible": license_compatible,
224        },
225        "check_details": {
226            "typosquatting": typosquat_check or "✓ No typosquatting detected",
227            "license": license_status,
228        },
229    }
230
231    # Check for suspicious indicators
232    risk_score = 0
233
234    # Typosquatting check
235    if typosquat_check:
236        checks["warnings"].append(typosquat_check)
237        risk_score += 5  # High risk
238
239    # License check
240    if not license_compatible:
241        checks["warnings"].append(f"License issue: {license_status}")
242        risk_score += 2
243
244    if age_days < 30:
245        checks["warnings"].append(f"Very new package (only {age_days} days old)")
246        risk_score += 3
247    elif age_days < 90:
248        checks["warnings"].append(f"Relatively new package ({age_days} days old)")
249        risk_score += 1
250
251    if checks["total_releases"] < 3:
252        checks["warnings"].append(f"Very few releases (only {checks['total_releases']})")
253        risk_score += 2
254
255    if not source:
256        checks["warnings"].append("No source repository linked")
257        risk_score += 2
258
259    if not homepage and not source:
260        checks["warnings"].append("No homepage or source repository")
261        risk_score += 1
262
263    if checks["author"] == "Unknown":
264        checks["warnings"].append("No author information available")
265        risk_score += 1
266
267    # Add informational items
268    checks["info_items"].append(f"Age: {age_days} days")
269    checks["info_items"].append(f"Releases: {checks['total_releases']}")
270    checks["info_items"].append(f"Author: {checks['author']}")
271    checks["info_items"].append(f"License: {checks['license']}")
272    if source:
273        checks["info_items"].append(f"Source: {source}")
274
275    # Determine risk level
276    if risk_score >= 5:
277        checks["risk_level"] = "high"
278    elif risk_score >= 3:
279        checks["risk_level"] = "medium"
280    else:
281        checks["risk_level"] = "low"
282
283    return checks
284
285
286def format_check_result(result: dict[str, Any]) -> str:
287    """
288    Format a check result for display.
289
290    :param result: The check result dictionary.
291    """
292    risk_emoji = {"high": "🔴", "medium": "🟡", "low": "🟢", "unknown": "⚪"}
293    version = result.get("version", "unknown")
294
295    lines = [f"\n{risk_emoji[result['risk_level']]} **{result['name']}** (v{version})"]
296
297    if result.get("error"):
298        lines.append(f"  ❌ {result['error']}")
299        return "\n".join(lines)
300
301    if result.get("summary"):
302        lines.append(f"  📝 {result['summary']}")
303
304    if result.get("info_items"):
305        for item in result["info_items"]:
306            lines.append(f"  ℹ️  {item}")
307
308    if result.get("warnings"):
309        for warning in result["warnings"]:
310            lines.append(f"  ⚠️  {warning}")
311
312    return "\n".join(lines)
313
314
315def main() -> int:
316    """Run the package safety check."""
317    if len(sys.argv) < 2:
318        print("Usage: check_package_safety.py <requirements_file_or_package_name>")
319        print("  Or: check_package_safety.py package1 package2 package3")
320        return 1
321
322    packages = []
323
324    # Check if first argument is a file
325    if len(sys.argv) == 2 and sys.argv[1].endswith(".txt"):
326        try:
327            with open(sys.argv[1]) as f:
328                for line in f:
329                    package = parse_requirement(line)
330                    if package:
331                        packages.append(package)
332        except FileNotFoundError:
333            print(f"Error: File '{sys.argv[1]}' not found")
334            return 1
335    else:
336        # Treat arguments as package names
337        packages = [arg.lower() for arg in sys.argv[1:]]
338
339    if not packages:
340        print("No packages to check")
341        return 0
342
343    print(f"Checking {len(packages)} package(s)...\n")
344    print("=" * 80)
345
346    results = []
347    for package in packages:
348        result = check_package(package)
349        results.append(result)
350        print(format_check_result(result))
351
352    print("\n" + "=" * 80)
353
354    # Automated checks summary
355    all_trusted = all(r.get("automated_checks", {}).get("trusted_source", False) for r in results)
356    all_no_typosquat = all(
357        r.get("automated_checks", {}).get("typosquatting", False) for r in results
358    )
359    all_license_ok = all(
360        r.get("automated_checks", {}).get("license_compatible", False) for r in results
361    )
362
363    print("\n🤖 Automated Security Checks:")
364    trusted_msg = (
365        "All packages have source repositories"
366        if all_trusted
367        else "Some packages missing source info"
368    )
369    print(f"  {'✅' if all_trusted else '❌'} Trusted Sources: {trusted_msg}")
370
371    typosquat_msg = (
372        "No suspicious package names detected"
373        if all_no_typosquat
374        else "Possible typosquatting detected!"
375    )
376    print(f"  {'✅' if all_no_typosquat else '❌'} Typosquatting: {typosquat_msg}")
377
378    license_msg = (
379        "All licenses are compatible" if all_license_ok else "Some license issues detected"
380    )
381    print(f"  {'✅' if all_license_ok else '⚠️ '} License Compatibility: {license_msg}")
382
383    # Summary
384    high_risk = sum(1 for r in results if r["risk_level"] == "high")
385    medium_risk = sum(1 for r in results if r["risk_level"] == "medium")
386    low_risk = sum(1 for r in results if r["risk_level"] == "low")
387
388    print(f"\n📊 Summary: {len(results)} packages checked")
389    if high_risk:
390        print(f"  🔴 High risk: {high_risk}")
391    if medium_risk:
392        print(f"  🟡 Medium risk: {medium_risk}")
393    print(f"  🟢 Low risk: {low_risk}")
394
395    if high_risk > 0:
396        print("\n⚠️  High-risk packages detected! Manual review strongly recommended.")
397        return 2
398    if medium_risk > 0:
399        print("\n⚠️  Medium-risk packages detected. Please review before merging.")
400        return 1
401
402    print("\n✅ All packages passed basic safety checks.")
403    return 0
404
405
406if __name__ == "__main__":
407    sys.exit(main())
408