Repository navigation
feat/0.16.0: web API analyze + compare endpoints #88
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
9716884
2864b94
5dc6235
80e0b83
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -763,6 +763,281 @@ def cache_clear() -> tuple[dict, int]: | |
| return jsonify({"cleared": count}), 200 | ||
|
|
||
|
|
||
| # ─── Analyze endpoint ────────────────────────────────────────── | ||
| @app.post("/api/v1/analyze") | ||
| @require_api_key | ||
| def analyze_post() -> tuple[dict, int]: | ||
| """ | ||
| Analyze package conflicts and dependencies. | ||
|
|
||
| Expected JSON body:: | ||
|
|
||
| { | ||
| "packages": ["pkg1", "pkg2"], | ||
| "manager": "apt", | ||
| "solver": "greedy", | ||
| "metadata": false, | ||
| "conflicts": [["pkg1", "pkg2"]], | ||
| "depends": [["pkg1", "dep1"]] | ||
| } | ||
| """ | ||
| from ..core.model_encoder import encode_packages | ||
| from ..core.package import Package | ||
|
|
||
| data = request.get_json(force=True) or {} | ||
| if not isinstance(data, dict): | ||
| return ( | ||
| jsonify( | ||
| { | ||
| "error": "Bad Request", | ||
| "message": "Request body must be a JSON object", | ||
| } | ||
| ), | ||
| 400, | ||
| ) | ||
|
|
||
| packages = data.get("packages", []) | ||
| if not isinstance(packages, list) or not packages: | ||
| return ( | ||
| jsonify( | ||
| { | ||
| "error": "Bad Request", | ||
| "message": "'packages' must be a non-empty list", | ||
| } | ||
| ), | ||
| 400, | ||
| ) | ||
|
|
||
| manager = data.get("manager", "apt") | ||
| solver = data.get("solver", "greedy") | ||
| metadata = data.get("metadata", False) | ||
| conflicts = data.get("conflicts", []) | ||
| depends = data.get("depends", []) | ||
|
|
||
| try: | ||
| manager_enum = PackageManagerType(manager) | ||
| except ValueError: | ||
| valid = [m.value for m in PackageManagerType] | ||
| return ( | ||
| jsonify( | ||
| { | ||
| "error": "Bad Request", | ||
| "message": f"Unknown manager '{manager}'. Valid: {valid}", | ||
| } | ||
| ), | ||
| 400, | ||
| ) | ||
|
|
||
| package_objs = [Package(name=p, status="candidate") for p in packages] | ||
|
|
||
| for conflict_pair in conflicts: | ||
| if len(conflict_pair) == 2: | ||
| a, b = conflict_pair | ||
| for pkg in package_objs: | ||
| if pkg.name == a: | ||
| if b not in pkg.conflicts: | ||
| pkg.conflicts.append(b) | ||
| if pkg.name == b: | ||
| if a not in pkg.conflicts: | ||
| pkg.conflicts.append(a) | ||
|
|
||
| for dep_pair in depends: | ||
| if len(dep_pair) == 2: | ||
| pkg_name, dep_name = dep_pair | ||
| for pkg in package_objs: | ||
| if pkg.name == pkg_name: | ||
| if dep_name not in pkg.depends: | ||
| pkg.depends.append(dep_name) | ||
|
|
||
| if metadata: | ||
| from ..adapters import get_adapter | ||
|
|
||
| try: | ||
| adapter = get_adapter(manager) | ||
| except Exception: | ||
| adapter = None | ||
|
|
||
| if adapter is not None: | ||
| for pkg in package_objs: | ||
| meta = adapter.fetch(pkg.name) | ||
| if meta and meta.name: | ||
| if meta.depends: | ||
| pkg.depends = list(dict.fromkeys(pkg.depends + meta.depends)) | ||
| if meta.conflicts: | ||
| pkg.conflicts = list( | ||
| dict.fromkeys(pkg.conflicts + meta.conflicts) | ||
| ) | ||
|
|
||
| constraints = encode_packages(package_objs) | ||
| maximizer = PackageMaximizer(manager=manager_enum, solver=solver) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win 🔎 Supported by static analysis🏁 Script executed: #!/bin/bash
# Confirm SolverType values and how PackageMaximizer handles unknown/non-string solver input.
fd -t f 'enums.py' package_maximizer --exec rg -n -A30 'class SolverType'
fd -t f 'maximizer.py' package_maximizer --exec rg -n -B3 -A20 'isinstance\(solver, str\)'
# Confirm the web module imports SolverType already, or whether a new import is needed.
fd -t f 'app.py' package_maximizer/web --exec rg -n '^(from|import) 'Repository: dominicusin/package-maximizer Length of output: 2816 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- app.py validation and call path ---'
sed -n '770,885p' package_maximizer/web/app.py
printf '%s\n' '--- PackageManagerType binding and constructor contract ---'
sed -n '1,90p' package_maximizer/core/enums.py
sed -n '45,80p' package_maximizer/core/maximizer.py
printf '%s\n' '--- solver extraction and relevant callers ---'
rg -n -C 3 'solver|manager_enum|PackageMaximizer\(' package_maximizer/web/app.pyRepository: dominicusin/package-maximizer Length of output: 17505 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- app.py validation and call path ---'
sed -n '770,885p' package_maximizer/web/app.py
printf '%s\n' '--- enum and constructor contract ---'
sed -n '1,90p' package_maximizer/core/enums.py
sed -n '45,80p' package_maximizer/core/maximizer.py
printf '%s\n' '--- solver extraction and relevant callers ---'
rg -n -C 3 'solver|manager_enum|PackageMaximizer\(' package_maximizer/web/app.pyRepository: dominicusin/package-maximizer Length of output: 17483 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- non-string solver path and solve receiver ---'
rg -n -C 8 'def _infer_solver_type|def solve\(' package_maximizer/core/maximizer.py
printf '%s\n' '--- solver registry keys ---'
rg -n -C 4 'SOLVER_REGISTRY' package_maximizer/solversRepository: dominicusin/package-maximizer Length of output: 3892 🏁 Script executed: #!/bin/bash
set -eu
sed -n '105,135p' package_maximizer/core/maximizer.py
sed -n '196,225p' package_maximizer/core/maximizer.py
sed -n '42,105p' package_maximizer/solvers/__init__.pyRepository: dominicusin/package-maximizer Length of output: 4468 Validate The + from ..core.enums import SolverType
+
+ if not isinstance(solver, str) or solver.lower() not in {
+ s.value for s in SolverType
+ }:
+ return (
+ jsonify(
+ {
+ "error": "Bad Request",
+ "message": f"Unknown solver '{solver}'. "
+ f"Valid: {[s.value for s in SolverType]}",
+ }
+ ),
+ 400,
+ )
+
constraints = encode_packages(package_objs)
maximizer = PackageMaximizer(manager=manager_enum, solver=solver)🤖 Prompt for AI Agents |
||
| result = maximizer.solve(package_objs) | ||
| selected_set = set(result) | ||
|
|
||
| all_names = {p.name for p in package_objs} | ||
| excluded = all_names - selected_set | ||
|
|
||
| conflict_map = {} | ||
| for name in sorted(excluded): | ||
| reasons = [] | ||
| for a, b in constraints.conflicts: | ||
| if a == name and b in selected_set: | ||
| reasons.append(f"conflict with selected '{b}'") | ||
| elif b == name and a in selected_set: | ||
| reasons.append(f"conflict with selected '{a}'") | ||
| deps = constraints.dependencies.get(name, []) | ||
| unmet_deps = [d for d in deps if d not in selected_set] | ||
| if unmet_deps: | ||
| reasons.append(f"unmet dependencies: {', '.join(unmet_deps)}") | ||
| conflict_map[name] = reasons if reasons else ["not selected"] | ||
|
|
||
| conflict_count = {} | ||
| for a, b in constraints.conflicts: | ||
| conflict_count[a] = conflict_count.get(a, 0) + 1 | ||
| conflict_count[b] = conflict_count.get(b, 0) + 1 | ||
| bottlenecks = sorted(conflict_count.items(), key=lambda x: x[1], reverse=True)[:5] | ||
|
|
||
| analysis = { | ||
| "total_packages": len(package_objs), | ||
| "selected_count": len(result), | ||
| "excluded_count": len(excluded), | ||
| "selected": result, | ||
| "excluded": {k: v for k, v in conflict_map.items() if v}, | ||
| "conflict_count": len(constraints.conflicts), | ||
| "dependency_count": sum(len(d) for d in constraints.dependencies.values()), | ||
| "bottlenecks": [{"name": n, "conflict_count": c} for n, c in bottlenecks], | ||
| "solver": solver, | ||
| "manager": manager, | ||
| } | ||
|
|
||
| return jsonify(analysis), 200 | ||
|
|
||
|
|
||
| # ─── Compare endpoint ────────────────────────────────────────── | ||
| @app.post("/api/v1/compare") | ||
| @require_api_key | ||
| def compare_post() -> tuple[dict, int]: | ||
| """ | ||
| Compare all available solvers on the given package set. | ||
|
|
||
| Expected JSON body:: | ||
|
|
||
| { | ||
| "packages": ["pkg1", "pkg2"], | ||
| "manager": "apt", | ||
| "metadata": false | ||
| } | ||
| """ | ||
| import time | ||
|
|
||
| from ..solvers import SOLVER_REGISTRY | ||
|
|
||
| data = request.get_json(force=True) or {} | ||
| if not isinstance(data, dict): | ||
| return ( | ||
| jsonify( | ||
| { | ||
| "error": "Bad Request", | ||
| "message": "Request body must be a JSON object", | ||
| } | ||
| ), | ||
| 400, | ||
| ) | ||
|
|
||
| packages = data.get("packages", []) | ||
| if not isinstance(packages, list) or not packages: | ||
| return ( | ||
| jsonify( | ||
| { | ||
| "error": "Bad Request", | ||
| "message": "'packages' must be a non-empty list", | ||
| } | ||
| ), | ||
| 400, | ||
| ) | ||
|
|
||
| manager = data.get("manager", "apt") | ||
| metadata = data.get("metadata", False) | ||
|
|
||
| try: | ||
| manager_enum = PackageManagerType(manager) | ||
| except ValueError: | ||
| valid = [m.value for m in PackageManagerType] | ||
| return ( | ||
| jsonify( | ||
| { | ||
| "error": "Bad Request", | ||
| "message": f"Unknown manager '{manager}'. Valid: {valid}", | ||
| } | ||
| ), | ||
| 400, | ||
| ) | ||
|
|
||
| package_objs = [Package(name=p, status="candidate") for p in packages] | ||
|
|
||
| if metadata: | ||
| from ..adapters import get_adapter | ||
|
|
||
| try: | ||
| adapter = get_adapter(manager) | ||
| except Exception: | ||
| adapter = None | ||
|
|
||
| if adapter is not None: | ||
| for pkg in package_objs: | ||
| pkg_meta = adapter.fetch(pkg.name) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🔒 Security & Privacy | 🛡️ Analyzed with Security Review | 🟠 Major | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
# Confirm user-supplied names reach subprocess argv without validation.
fd -t f '__init__.py' package_maximizer/adapters --exec rg -n -B4 -A12 'subprocess\.run'Repository: dominicusin/package-maximizer Length of output: 3475 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- analyze_post input and enrichment path ---'
sed -n '770,885p' package_maximizer/web/app.py
printf '%s\n' '--- adapter definitions and fetch calls ---'
sed -n '1,210p' package_maximizer/adapters/__init__.py
sed -n '250,375p' package_maximizer/adapters/__init__.py
sed -n '440,475p' package_maximizer/adapters/__init__.pyRepository: dominicusin/package-maximizer Length of output: 15863 SSRF (CWE-88): Improper Neutralization of Argument Delimiters in a Command ('Argument Injection') Reachability: External · Exploitability: Moderate Reject option-like package names before metadata lookup.
🤖 Prompt for AI AgentsSource: Linters/SAST tools |
||
| if pkg_meta and pkg_meta.name: | ||
| if pkg_meta.depends: | ||
| pkg.depends = list(pkg_meta.depends) | ||
| if pkg_meta.conflicts: | ||
| pkg.conflicts = list(pkg_meta.conflicts) | ||
|
|
||
| solver_names = list(SOLVER_REGISTRY.keys()) | ||
| results = [] | ||
|
|
||
| for sname in solver_names: | ||
| try: | ||
| solver_cls = SOLVER_REGISTRY[sname] | ||
| solver_inst = solver_cls() | ||
| start = time.time() | ||
| res = solver_inst.solve(list(package_objs)) | ||
| elapsed = time.time() - start | ||
| results.append( | ||
| { | ||
| "solver": sname, | ||
| "avg_time": elapsed, | ||
| "selected_count": len(res), | ||
| "selected": res, | ||
| "success": True, | ||
| "error": None, | ||
| } | ||
| ) | ||
| except Exception as e: # noqa: BLE001 | ||
| results.append( | ||
| { | ||
| "solver": sname, | ||
| "avg_time": 0.0, | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win A failed solver becomes The failure branch records Order failures last and select the best solver from successful results only. 🐛 Proposed fix- results.sort(key=lambda r: r["avg_time"])
+ results.sort(key=lambda r: (not r["success"], r["avg_time"]))
return (
jsonify(
{
"results": results,
"package_count": len(packages),
- "best_solver": results[0]["solver"] if results else None,
+ "best_solver": next(
+ (r["solver"] for r in results if r["success"]), None
+ ),
"manager": manager,
}
),
200,
)Also applies to: 1026-1026, 1033-1033 🤖 Prompt for AI Agents |
||
| "selected_count": 0, | ||
| "selected": [], | ||
| "success": False, | ||
| "error": str(e), | ||
| } | ||
| ) | ||
|
|
||
| results.sort(key=lambda r: r["avg_time"]) | ||
|
|
||
| return ( | ||
| jsonify( | ||
| { | ||
| "results": results, | ||
| "package_count": len(packages), | ||
| "best_solver": results[0]["solver"] if results else None, | ||
| "manager": manager, | ||
| } | ||
| ), | ||
| 200, | ||
| ) | ||
|
|
||
|
|
||
| # ─── Error handlers ────────────────────────────────────────── | ||
| @app.errorhandler(404) | ||
| def not_found(e): | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Validate the types of
packages,conflicts, anddependsbefore use.The endpoint checks only that
packagesis a non-empty list. Malformed but valid JSON then reaches code that raises an unhandled exception, and Flask returns 500 instead of 400.Concrete cases:
{"packages": ["a"], "conflicts": null}— Line 833 iteratesNoneand raisesTypeError.{"packages": ["a"], "depends": 5}— Line 844 iterates anintand raisesTypeError.{"packages": [{"x": 1}]}— Line 876 builds a set from unhashabledictnames and raisesTypeError.Reject these inputs with 400.
🛡️ Proposed validation
packages = data.get("packages", []) - if not isinstance(packages, list) or not packages: + if ( + not isinstance(packages, list) + or not packages + or not all(isinstance(p, str) and p for p in packages) + ): return ( jsonify( { "error": "Bad Request", - "message": "'packages' must be a non-empty list", + "message": "'packages' must be a non-empty list of strings", } ), 400, )conflicts = data.get("conflicts", []) depends = data.get("depends", []) + + def _valid_pairs(value) -> bool: + return isinstance(value, list) and all( + isinstance(pair, (list, tuple)) + and len(pair) == 2 + and all(isinstance(x, str) for x in pair) + for pair in value + ) + + if not _valid_pairs(conflicts) or not _valid_pairs(depends): + return ( + jsonify( + { + "error": "Bad Request", + "message": "'conflicts' and 'depends' must be lists of [name, name] pairs", + } + ), + 400, + )Also applies to: 833-845
🤖 Prompt for AI Agents