#!/usr/bin/env python3
"""Check an Android build with APKreport from a pipeline (#158).

    APKREPORT_KEY=apk_... python3 apkreport-check.py app-release.apk --fail-below 70 \\
        --fail-on new-tracker,new-dangerous-permission --sarif apkreport.sarif --comment github

The build goes up in 48 MB slices, as the site's own uploader sends it, so no request is larger than
the site takes in one and none waits for the report: the job is asked after until it is through.
Then the site sets the build against the build before it of the same app from the same account or
team (/api/artifacts/<sha256>/check) and says pass or fail by the thresholds given. The verdict is the
server's; this script prints it, writes the summary and the SARIF, posts or updates one comment on the
pull request, and exits with it.

Exit status: 0 pass, 1 fail by a threshold, 2 the build could not be checked.
Python 3.8 or later, the standard library alone. The docs: https://apkreport.com/docs/ci
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

SLICE = 48 * 1024 * 1024
MARKER = "<!-- apkreport-check -->"
DEFAULT_URL = "https://apkreport.com"
PASS, FAIL, ERROR = 0, 1, 2


class Failed(Exception):
    """The build could not be checked; the message says why."""


def urllib_transport(method: str, url: str, headers: dict, body: bytes | None, timeout: float):
    """(status, headers, body) of one request; an HTTP error is an answer too, not an exception."""
    request = urllib.request.Request(url, data=body, method=method, headers=headers)
    try:
        with urllib.request.urlopen(request, timeout=timeout) as answer:
            return answer.status, {k.lower(): v for k, v in answer.headers.items()}, answer.read()
    except urllib.error.HTTPError as exc:
        return exc.code, {k.lower(): v for k, v in exc.headers.items()}, exc.read()


# Swapped for a test's; every request of this script goes through it.
TRANSPORT = urllib_transport


def call(method: str, url: str, headers: dict | None = None, body=None, timeout: float = 900.0,
         tries: int = 5, expect=(200, 201, 202)):
    """One request, tried again on a lost connection, a 429 (after its Retry-After) and a 5xx.
    A dict body goes as JSON. The answer's JSON, or its bytes when it is not JSON."""
    headers = dict(headers or {})
    if isinstance(body, (dict, list)):
        body = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    headers.setdefault("User-Agent", "apkreport-check/1")
    for attempt in range(1, tries + 1):
        try:
            status, answer_headers, data = TRANSPORT(method, url, headers, body, timeout)
        except (OSError, urllib.error.URLError) as exc:
            if attempt == tries:
                raise Failed(f"{method} {url}: {exc}") from exc
            time.sleep(min(30, 2 ** attempt))
            continue
        if status in expect:
            try:
                return json.loads(data or b"null")
            except ValueError:
                return data
        if (status == 429 or status >= 500) and attempt < tries:
            wait = answer_headers.get("retry-after", "")
            time.sleep(min(120, int(wait)) if wait.isdigit() else min(30, 2 ** attempt))
            continue
        raise Failed(f"{method} {url.split('?')[0]}: {status} {detail_of(data)}", status, answer_headers)
    raise Failed(f"{method} {url}: no answer")


def detail_of(data: bytes) -> str:
    try:
        answer = json.loads(data)
        return str(answer.get("detail") if isinstance(answer, dict) else answer)
    except ValueError:
        return (data or b"")[:200].decode("utf-8", "replace").strip()


class Site:
    def __init__(self, url: str, key: str):
        self.url = url.rstrip("/")
        self.auth = {"Authorization": f"Bearer {key}"}

    def __call__(self, method: str, path: str, body=None, **options):
        return call(method, self.url + path, self.auth, body, **options)

    def send(self, paths: list[str], keep: bool = False) -> dict:
        """The files up in slices and handed over: the answer of /done - a job, or the build on file."""
        sizes = [os.path.getsize(path) for path in paths]
        started = self("POST", "/api/uploads", {"files": [os.path.basename(p) for p in paths], "sizes": sizes})
        upload = started["id"]
        for index, path in enumerate(paths):
            with open(path, "rb") as file:
                offset = 0
                while offset < sizes[index]:
                    file.seek(offset)
                    chunk = file.read(SLICE)
                    try:
                        self("POST", f"/api/uploads/{upload}/parts/{index}?at={offset}", chunk)
                    except Failed as exc:
                        # The part holds other than what was sent: go on from where the site says it is.
                        held = exc.args[2].get("x-part-size", "") if len(exc.args) > 2 and exc.args[1] == 409 else ""
                        if not held.isdigit():
                            raise
                        offset = int(held)
                        continue
                    offset += len(chunk)
        return self("POST", f"/api/uploads/{upload}/done", {"keep": keep, "source": "upload"})

    def wait(self, answer: dict, timeout: float) -> str:
        """The build's digest once its report is there."""
        if answer.get("existing"):
            return answer["existing"]
        job, until, state = answer.get("job"), time.monotonic() + timeout, "queued"
        while time.monotonic() < until:
            jobs = self("GET", f"/api/jobs?ids={urllib.parse.quote(job)}").get("jobs") or []
            state = jobs[0].get("state") if jobs else "gone"
            if state == "done":
                return (jobs[0].get("sha256") or "").split(",")[0]
            if state == "failed":
                raise Failed(f"the site could not read the build: {jobs[0].get('error') or 'no reason given'}")
            if state == "gone":
                raise Failed(f"job {job} is not known to the site")
            time.sleep(5)
        raise Failed(f"no report after {int(timeout)} s; job {job} is still {state}")


def query(**values) -> str:
    pairs = [(k, v) for k, v in values.items() if v not in (None, "", [])]
    return "?" + urllib.parse.urlencode(pairs, doseq=True) if pairs else ""


def signed(value) -> str:
    return f"+{value}" if value and value > 0 else str(value)


def markdown(check: dict) -> str:
    """The pull request's comment and the job's summary: the score, what changed, the verdict."""
    score, verdict, previous = check["score"], check["verdict"], check.get("previous")
    name = check.get("label") or check.get("package") or check["sha256"][:12]
    version = check.get("version_name") or check.get("version_code") or "?"
    lines = [MARKER, f"### APKreport: {name} {version} scores {score['total']} ({score['band']})", ""]
    if previous:
        was = previous.get("version_name") or previous.get("version_code") or previous["sha256"][:12]
        delta = check.get("score_delta") or 0
        side = "a newer build" if previous.get("against_newer") else "the build before"
        lines.append(f"Against {side}, {was} ({previous['score']['total']}): "
                     f"{'no change in the score' if not delta else signed(delta)}.")
    else:
        lines.append("The first build of this app from your account or team: nothing to compare with yet.")
    lines.append("")
    verdict_line = "**Pass.**" if verdict["pass"] else "**Fail.**"
    lines.append(verdict_line + "".join(f" {reason}" for reason in verdict["reasons"]))
    for note in verdict.get("notes") or []:
        lines.append(f"_{note}_")
    if previous:
        rows = []
        for title, items in (("New trackers", check["trackers"]["new"]),
                             ("New dangerous permissions", check["dangerous_permissions"]["new"]),
                             ("Trackers gone", check["trackers"]["gone"]),
                             ("Dangerous permissions gone", check["dangerous_permissions"]["gone"])):
            if items:
                rows.append(f"| {title} | {', '.join(f'`{i}`' for i in items)} |")
        if check["axes_delta"]:
            rows.append("| Axes | " + ", ".join(f"{k} {signed(v)}" for k, v in check["axes_delta"].items()) + " |")
        if rows:
            lines += ["", "| | |", "|---|---|"] + rows
        if check["findings"]["new"]:
            lines += ["", "New findings:"]
            lines += [f"- **{f['severity']}** {f['title']}" + (f" (was {f['was']})" if f.get("was") else "")
                      for f in check["findings"]["new"]]
        if check["findings"]["resolved"]:
            lines += ["", "Resolved: " + "; ".join(f["title"] for f in check["findings"]["resolved"]) + "."]
        if check.get("changes") and check["changes"].get("summary"):
            lines += ["", f"Changes: {check['changes']['summary']}."]
    lines += ["", f"[The report]({check['url']}) opens to your account and your team."]
    return "\n".join(lines) + "\n"


# --- the pull request's comment ---------------------------------------------------------------

def event_number() -> int | None:
    """The pull request a GitHub or Gitea job runs for, from its event."""
    path = os.environ.get("GITHUB_EVENT_PATH", "")
    if not path or not os.path.exists(path):
        return None
    with open(path, encoding="utf-8") as file:
        event = json.load(file)
    number = (event.get("pull_request") or {}).get("number") or (event.get("number") if "pull_request" in event else None)
    return int(number) if number else None


def comment(kind: str, body: str) -> str:
    """Post the comment, or update the one this script posted before: one comment a pull request."""
    if kind == "gitlab":
        api, project, mr = (os.environ.get(k, "") for k in ("CI_API_V4_URL", "CI_PROJECT_ID", "CI_MERGE_REQUEST_IID"))
        token = os.environ.get("APKREPORT_GITLAB_TOKEN") or os.environ.get("GITLAB_TOKEN", "")
        if not (api and project and mr):
            return "not a merge request pipeline: no comment"
        if not token:
            return "no APKREPORT_GITLAB_TOKEN: no comment"
        auth = {"PRIVATE-TOKEN": token}
        notes = f"{api}/projects/{urllib.parse.quote(project, safe='')}/merge_requests/{mr}/notes"
        for note in call("GET", notes + "?per_page=100&sort=asc", auth) or []:
            if MARKER in (note.get("body") or ""):
                call("PUT", f"{notes}/{note['id']}", auth, {"body": body})
                return f"comment updated on merge request !{mr}"
        call("POST", notes, auth, {"body": body})
        return f"comment posted on merge request !{mr}"
    number = event_number()
    repo = os.environ.get("GITHUB_REPOSITORY", "")
    if number is None or not repo:
        return "not a pull request: no comment"
    if kind == "gitea":
        api = os.environ.get("GITHUB_API_URL") or os.environ.get("GITHUB_SERVER_URL", "").rstrip("/") + "/api/v1"
        token = os.environ.get("APKREPORT_GITEA_TOKEN") or os.environ.get("GITEA_TOKEN") or os.environ.get("GITHUB_TOKEN", "")
        auth = {"Authorization": f"token {token}"}
    else:
        api = os.environ.get("GITHUB_API_URL", "https://api.github.com")
        token = os.environ.get("GITHUB_TOKEN", "")
        auth = {"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"}
    if not token:
        return "no token for the comment: no comment"
    api = api.rstrip("/")
    for page in range(1, 11):
        found = call("GET", f"{api}/repos/{repo}/issues/{number}/comments?per_page=100&page={page}", auth) or []
        for item in found:
            if MARKER in (item.get("body") or ""):
                call("PATCH", f"{api}/repos/{repo}/issues/comments/{item['id']}", auth, {"body": body})
                return f"comment updated on pull request #{number}"
        if len(found) < 100:
            break
    call("POST", f"{api}/repos/{repo}/issues/{number}/comments", auth, {"body": body})
    return f"comment posted on pull request #{number}"


def output(**values) -> None:
    """The step's outputs, for a GitHub or Gitea job's next steps."""
    path = os.environ.get("GITHUB_OUTPUT", "")
    if path:
        with open(path, "a", encoding="utf-8") as file:
            for key, value in values.items():
                file.write(f"{key}={value}\n")


def arguments(argv=None):
    parser = argparse.ArgumentParser(description="Check an Android build with APKreport.")
    parser.add_argument("files", nargs="+", help="the APK, or a base APK and its splits, or one .apks/.xapk/.apkm")
    parser.add_argument("--url", default=os.environ.get("APKREPORT_URL", DEFAULT_URL))
    parser.add_argument("--fail-below", type=int, default=None, metavar="SCORE")
    parser.add_argument("--fail-on", action="append", default=[], metavar="WHAT",
                        help="new-tracker, new-dangerous-permission, new-finding, score-drop; comma-separated or repeated")
    parser.add_argument("--against", default="", metavar="SHA256", help="the build to compare with, by its digest")
    parser.add_argument("--keep", action="store_true", help="keep the file for as long as the plan says")
    parser.add_argument("--sarif", default="", metavar="PATH", help="write the findings as SARIF here")
    parser.add_argument("--sarif-uri", default="", metavar="PATH", help="where the alerts point: the APK's path in the repository")
    parser.add_argument("--manifest", default="", metavar="PATH", help="the sources' AndroidManifest.xml, for what it says")
    parser.add_argument("--comment", choices=("github", "gitea", "gitlab"), default=None)
    parser.add_argument("--summary", default=os.environ.get("GITHUB_STEP_SUMMARY", ""), metavar="PATH")
    parser.add_argument("--json", default="", metavar="PATH", help="write the site's answer here")
    parser.add_argument("--timeout", type=float, default=1800.0, metavar="SECONDS")
    parser.add_argument("--no-fail", action="store_true", help="report, and exit 0 whatever the verdict")
    return parser.parse_args(argv)


def main(argv=None) -> int:
    args = arguments(argv)
    key = os.environ.get("APKREPORT_KEY", "")
    if not key:
        print("apkreport: APKREPORT_KEY is not set; make a key on your account page", file=sys.stderr)
        return ERROR
    site = Site(args.url, key)
    try:
        for path in args.files:
            if not os.path.isfile(path):
                raise Failed(f"{path}: no such file")
        sha256 = site.wait(site.send(args.files, keep=args.keep), args.timeout)
        check = site("GET", f"/api/artifacts/{sha256}/check" + query(
            fail_below=args.fail_below, fail_on=args.fail_on, against=args.against))
        if args.sarif:
            uri = args.sarif_uri or os.path.relpath(args.files[0]).replace(os.sep, "/")
            document = site("GET", f"/api/artifacts/{sha256}/sarif" + query(uri=uri, manifest=args.manifest))
            with open(args.sarif, "w", encoding="utf-8") as file:
                json.dump(document, file, indent=2)
    except Failed as exc:
        print(f"apkreport: {exc.args[0]}", file=sys.stderr)
        output(verdict="error")
        return ERROR
    body = markdown(check)
    print(body.replace(MARKER + "\n", ""))
    if args.json:
        with open(args.json, "w", encoding="utf-8") as file:
            json.dump(check, file, indent=2)
    if args.summary:
        with open(args.summary, "a", encoding="utf-8") as file:
            file.write(body)
    if args.comment:
        try:
            print(f"apkreport: {comment(args.comment, body)}")
        except Failed as exc:
            # A comment that could not be posted fails nothing: the verdict is the job's.
            print(f"apkreport: the comment was not posted: {exc.args[0]}", file=sys.stderr)
    verdict = check["verdict"]
    output(sha256=sha256, score=check["score"]["total"], band=check["score"]["band"], verdict=verdict["state"],
           url=check["url"], sarif=args.sarif)
    if not verdict["pass"] and not args.no_fail:
        for reason in verdict["reasons"]:
            print(f"apkreport: fail: {reason}", file=sys.stderr)
        return FAIL
    return PASS


if __name__ == "__main__":
    sys.exit(main())