Pewarta Proxy

📋 Server Request & Proxy Logs

Loading active proxy settings...
Developed by Khairul Adha (adhexa.id)
Mengambil log server...

⚙️ Pengaturan Proxy Target

Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions .github/workflows/build_wppm_prod_publish.yml
Original file line number Diff line number Diff line change
Expand Up @@ -91,3 +91,23 @@ jobs:
with:
skip-existing: true
verbose: true # Enable verbose output

tag_release:
name: Tag the published commit
needs: [upload_pypi_test]
runs-on: ubuntu-24.04
permissions:
contents: write
steps:
- uses: actions/checkout@v6

- name: tag wppm-<version>, once
shell: bash
run: |
TAG="wppm-$(sed -n "s/^__version__ = '\(.*\)'/\1/p" wppm/__init__.py)"
if git ls-remote --exit-code --tags origin "refs/tags/$TAG" >/dev/null; then
echo "$TAG already exists"
else
git tag "$TAG" "$GITHUB_SHA"
git push origin "$TAG"
fi
15 changes: 15 additions & 0 deletions README_PYPI.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,21 @@ one. No distribution is involved, so two downloaded files are enough. WinPython
publishes one index per release and flavor, in
[changelogs/](https://github.com/winpython/winpython/tree/master/changelogs).

## Security advisories

`-sa` asks [osv.dev](https://osv.dev) about the installed packages, or about the pins of
a `pylock.toml` or `requirements.txt`. One line per affected package, with the version
that fixes all its known advisories:

```console
$ wppm pylock.toml -sa
mako 1.3.12 -> 1.4.2 GHSA-5639-2j2p-m4mx
nbconvert 7.17.1 -> 7.17.2 GHSA-pjvx-prjx-h659
```

`-v` adds each advisory's summary, `-j` gives JSON. The exit code is 1 when anything is
found, so it can gate a build. It needs network access.

## Environments you have not installed anything into

`-t` points `wppm` at *another* Python distribution, and `-ws` at a plain directory of
Expand Down
40 changes: 40 additions & 0 deletions tests/test_audit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
# -*- coding: utf-8 -*-
"""wppm -sa reads pins from files offline; only the osv.dev query needs network."""
from wppm import audit


def test_pins_from_requirements(tmp_path):
req = tmp_path / "requirements.txt"
req.write_text("# comment\nMako==1.3.12\nnbconvert[webpdf] == 7.17.1 ; python_version>'3'\nnumpy>=2\n", encoding="utf-8")
assert audit.pins_from_file(req) == [("Mako", "1.3.12"), ("nbconvert", "7.17.1")]


def test_pins_from_pylock(tmp_path):
lock = tmp_path / "pylock.toml"
lock.write_text('lock-version = "1.0"\n[[packages]]\nname = "mako"\nversion = "1.3.12"\n'
'[[packages]]\nname = "local"\n', encoding="utf-8")
assert audit.pins_from_file(lock) == [("mako", "1.3.12")]


def test_fixed_in_picks_lowest_fix_above_installed():
vuln = {"affected": [{"package": {"ecosystem": "PyPI", "name": "Pydantic_AI-slim"},
"ranges": [{"type": "ECOSYSTEM", "events": [{"introduced": "0"}, {"fixed": "1.0"}]},
{"type": "ECOSYSTEM", "events": [{"introduced": "1.100"}, {"fixed": "1.107.4"}]},
{"type": "GIT", "events": [{"fixed": "01e0558825b8"}]}]}]}
assert audit._fixed_in(vuln, "pydantic-ai-slim", "1.106.0") == "1.107.4"
assert audit._fixed_in(vuln, "other", "1.106.0") == ""


def test_upgrades_flags_a_fix_with_advisories_of_its_own(monkeypatch):
rows = [{"package": "a", "version": "1.0", "fixed": "1.2"}, {"package": "a", "version": "1.0", "fixed": "1.10"},
{"package": "b", "version": "2.0", "fixed": "2.1"}, {"package": "c", "version": "3.0", "fixed": ""}]
asked = []
def fake_audit(pins):
asked.extend(pins)
return [{"package": "b"}]
monkeypatch.setattr(audit, "audit", fake_audit)
result = audit.upgrades(rows)
assert sorted(asked) == [("a", "1.10"), ("b", "2.1")]
assert result == {"a": {"version": "1.0", "to": "1.10", "clean": True},
"b": {"version": "2.0", "to": "2.1", "clean": False},
"c": {"version": "3.0", "to": "", "clean": None}}
2 changes: 1 addition & 1 deletion wppm/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,6 @@
OTHER DEALINGS IN THE SOFTWARE.
"""

__version__ = '17.13.20261027'
__version__ = '17.14.20261027'
__license__ = __doc__
__project_url__ = 'http://winpython.github.io/'
121 changes: 121 additions & 0 deletions wppm/audit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
# -*- coding: utf-8 -*-
"""Known security advisories for a set of pinned packages, from osv.dev.

One batch query covers the whole set; details are fetched only for the hits.
Needs network access, no key.
"""
import json
import re
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

try:
from packaging.version import Version
except ImportError:
from pip._vendor.packaging.version import Version

OSV = "https://api.osv.dev/v1"
TIMEOUT = 30
BATCH = 1000 # osv.dev limit per querybatch


def _call(url, payload=None):
data = json.dumps(payload).encode() if payload is not None else None
request = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
return json.load(response)


def pins_from_file(path):
"""(name, version) pairs of a pylock.toml or of the `name==version` lines of a requirements file."""
path = Path(path)
text = path.read_text(encoding="utf-8")
if path.suffix == ".toml":
try:
import tomllib
except ImportError:
import tomli as tomllib
return [(p["name"], p["version"]) for p in tomllib.loads(text).get("packages", []) if p.get("version")]
pins = []
for line in text.splitlines():
match = re.match(r"\s*([A-Za-z0-9][A-Za-z0-9._-]*)(?:\[[^\]]*\])?\s*===?\s*([^\s;#\\]+)", line)
if match:
pins.append(match.groups())
return pins


def _fixed_in(vuln, name, installed):
"""Lowest version that fixes `vuln` above `installed`, '' when none is known."""
normal = re.sub(r"[-_.]+", "-", name).lower()
fixes = []
for affected in vuln.get("affected", []):
package = affected.get("package", {})
if package.get("ecosystem") != "PyPI" or re.sub(r"[-_.]+", "-", package.get("name", "")).lower() != normal:
continue
for span in affected.get("ranges", []):
for event in span.get("events", []):
try:
if "fixed" in event and Version(event["fixed"]) > Version(installed):
fixes.append(Version(event["fixed"]))
except Exception:
pass
return str(min(fixes)) if fixes else ""


def audit(pins):
"""Advisories per pinned package: [{package, version, id, aliases, fixed, summary}], sorted.

The same flaw is often published twice (PYSEC and GHSA, aliases of each
other): it is reported once. A PYSEC record whose GHSA alias does not match
the same version is dropped: PYSEC often marks every version affected when
only a git commit is known as the fix, GHSA carries the curated range.
"""
pins = sorted(set(pins), key=lambda p: p[0].lower())
hits = []
for start in range(0, len(pins), BATCH):
chunk = pins[start:start + BATCH]
queries = [{"package": {"name": n, "ecosystem": "PyPI"}, "version": v} for n, v in chunk]
answers = _call(f"{OSV}/querybatch", {"queries": queries}).get("results", [])
for (name, version), answer in zip(chunk, answers):
hits += [(name, version, v["id"]) for v in answer.get("vulns") or []]
with ThreadPoolExecutor(8) as pool:
details = dict(zip((h[2] for h in hits), pool.map(lambda h: _call(f"{OSV}/vulns/{h[2]}"), hits)))
rows, seen = [], set()
for name, version, vid in sorted(hits, key=lambda h: (h[0].lower(), not h[2].startswith("GHSA"), h[2])):
vuln = details[vid]
aliases = [a for a in vuln.get("aliases", []) if a != vid]
if (name, vid) in seen:
continue
if vid.startswith("PYSEC") and any(a.startswith("GHSA") for a in aliases):
continue # its GHSA alias was not a hit, or it would be in seen
seen.update((name, a) for a in [vid] + aliases)
rows.append({"package": name, "version": version, "id": vid, "aliases": aliases,
"fixed": _fixed_in(vuln, name, version),
"summary": vuln.get("summary") or (vuln.get("details") or "").split("\n")[0]})
return rows


def upgrades(rows):
"""Per affected package, the version fixing all its advisories, and whether it is clean.

The highest `fixed` can carry an advisory of its own (pypdf needed eleven
releases to get past that), so the suggested versions are audited too: one
more batch query. Not clean means a later release is needed; it is not
searched for. {package: {"version", "to", "clean"}}, `to` '' and `clean`
None when some advisory has no known fix.
"""
result = {}
for row in rows:
entry = result.setdefault(row["package"], {"version": row["version"], "fixes": []})
entry["fixes"].append(row["fixed"])
for entry in result.values():
fixes = entry.pop("fixes")
entry["to"] = max(fixes, key=Version) if all(fixes) else ""
entry["clean"] = None
proposed = [(name, entry["to"]) for name, entry in result.items() if entry["to"]]
still = {row["package"] for row in audit(proposed)} if proposed else set()
for name, entry in result.items():
if entry["to"]:
entry["clean"] = name not in still
return result
24 changes: 24 additions & 0 deletions wppm/wppm.py
Original file line number Diff line number Diff line change
Expand Up @@ -351,6 +351,7 @@ def main(test=False):
parser.add_argument("-r", dest="pipup", action="store_true", help=f"show Reverse (!= constraining) dependancies of the given package[option]: wppm -r pytest![test]")
parser.add_argument("-tl", "--top-level", action="store_true", help="keep only the entries no other entry pulls in, sorted: wppm -tl, wppm requirements.txt -tl -v\nwith -p or -r, start the tree from them instead of every package: wppm -tl -p")
parser.add_argument("-l", dest="levels", type=int, default=-1, help="show 'LEVELS' levels of dependencies (with -p, -r): wppm -p pandas -l1")
parser.add_argument("-sa", "--audit", action="store_true", help="show known security advisories (osv.dev) of installed packages, or of a pylock.toml / requirements.txt: wppm -sa, wppm pylock.toml -sa")
parser.add_argument("-j", "--json", dest="json", action="store_true", help="machine-readable JSON output (with -p, -r, -ls, -md, -tl): wppm -p pandas[.] -j")
parser.add_argument("-t", dest="target", default=sys.prefix, help=f'path to target Python distribution (default: "{sys.prefix}")')
parser.add_argument("-i", "--install", action="store_true", help="install a given package wheel or pylock file (use pip for more features)")
Expand Down Expand Up @@ -379,6 +380,29 @@ def main(test=False):
raise RuntimeError("Incompatible arguments: --install and --uninstall")
if args.registerWinPython and args.unregisterWinPython:
raise RuntimeError("Incompatible arguments: --install and --uninstall")
if args.audit:
from . import audit
source = next((Path(f) for f in args.fname if f and Path(f).is_file()), None)
pins = audit.pins_from_file(source) if source else piptree.PipData(targetpython, args.wheelsource).pip_list()
rows = audit.audit(pins)
upgrades = audit.upgrades(rows)
if args.json:
print(json.dumps([dict(r, to=upgrades[r["package"]]["to"], clean=upgrades[r["package"]]["clean"]) for r in rows], indent=4))
else:
# one line per package, with the version fixing all its known advisories; -v lists them
for package, up in upgrades.items():
own = [r for r in rows if r["package"] == package]
if not up["to"]:
fix = " (no fix known for all)"
else:
fix = f" -> {up['to']}" + ("" if up["clean"] else " (not clean yet: has advisories of its own)")
ids = ", ".join(r["id"] for r in own) if len(own) <= 3 else f"{len(own)} advisories"
print(f"{package} {up['version']}{fix} {ids}")
if args.verbose:
for r in own:
print(f" {r['id']} -> {r['fixed'] or '?'} {utils.sum_up(r['summary'], 90)}")
print(f"# {len(pins)} packages checked, {len(rows)} advisories in {len({r['package'] for r in rows})} packages", file=sys.stderr)
sys.exit(1 if rows else 0)
if args.pipdown or args.pipup:
pip = piptree.PipData(targetpython, args.wheelsource)
# -tl is then a root filter, not a list of its own: "." asked for implicitly,
Expand Down
Loading