119 lines
5.4 KiB
Python
119 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Check one stored GCM PAT without asking Git to erase or replace it."""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import unquote, urlsplit
|
|
from urllib.request import HTTPRedirectHandler, Request, build_opener
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from credential_migrate import GitHelper, MigrationError, checked_request, safe_environment
|
|
|
|
|
|
class ProbeError(Exception):
|
|
pass
|
|
|
|
|
|
class NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
def git_read_url(site, remote):
|
|
try:
|
|
expected = checked_request(site, "account")["host"]
|
|
parts = urlsplit(remote)
|
|
if (parts.scheme.lower() != "https" or not parts.hostname or parts.username is not None
|
|
or parts.password is not None or parts.query or parts.fragment):
|
|
raise ValueError()
|
|
actual = checked_request("https://" + parts.netloc, "account")["host"]
|
|
path = unquote(parts.path)
|
|
if (actual != expected or not path.endswith(".git") or ".." in path.split("/")
|
|
or any(ord(c) < 33 or ord(c) == 127 for c in path)):
|
|
raise ValueError()
|
|
return site.rstrip("/") + parts.path + "/info/refs?service=git-upload-pack"
|
|
except (ValueError, UnicodeError, MigrationError):
|
|
raise ProbeError("The remote is not a clean repository URL on the selected HTTPS site.") from None
|
|
|
|
|
|
def load_credential(site, account, helper):
|
|
try:
|
|
request = checked_request(site, account)
|
|
result = helper.get(request)
|
|
if (not result or result.get("username") != request["username"] or not result.get("password")):
|
|
raise ProbeError("No matching complete credential was returned by GCM.")
|
|
for field in ("protocol", "host"):
|
|
if field in result and result[field] != request[field]:
|
|
raise ProbeError("The GCM credential site does not match.")
|
|
token = result["password"]
|
|
decoded = token.decode("ascii")
|
|
if any(ord(c) < 33 or ord(c) > 126 for c in decoded):
|
|
raise ValueError()
|
|
return token
|
|
except ProbeError:
|
|
raise
|
|
except (MigrationError, UnicodeError, ValueError):
|
|
raise ProbeError("The stored credential is unavailable or unsuitable for a PAT check.") from None
|
|
|
|
|
|
def check_requests(site, account, token, remote, opener):
|
|
git_url = git_read_url(site, remote)
|
|
token_text = token.decode("ascii")
|
|
api_request = Request(site.rstrip("/") + "/api/v1/user", headers={"Authorization": "token " + token_text, "Accept": "application/json"})
|
|
basic = base64.b64encode(account.encode("utf-8") + b":" + token).decode("ascii")
|
|
git_request = Request(git_url, headers={"Authorization": "Basic " + basic, "Accept": "application/x-git-upload-pack-advertisement"})
|
|
try:
|
|
with opener.open(api_request, timeout=15) as response:
|
|
data = json.loads(response.read(65536))
|
|
api_status = ("matched_account" if isinstance(data, dict) and str(data.get("login", "")).casefold() == account.casefold()
|
|
else "identity_mismatch_or_unexpected_response")
|
|
except HTTPError as exc:
|
|
api_status = "http_" + str(exc.code)
|
|
except (URLError, TimeoutError, OSError, ValueError):
|
|
api_status = "network_tls_or_invalid_response"
|
|
try:
|
|
with opener.open(git_request, timeout=15) as response:
|
|
git_status = ("read_advertisement" if response.headers.get_content_type() == "application/x-git-upload-pack-advertisement"
|
|
else "unexpected_response")
|
|
except HTTPError as exc:
|
|
git_status = "http_" + str(exc.code)
|
|
except (URLError, TimeoutError, OSError, ValueError):
|
|
git_status = "network_tls_or_invalid_response"
|
|
return api_status, git_status
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repo", required=True, help="local repository path")
|
|
parser.add_argument("--site", required=True, help="confirmed HTTPS site origin")
|
|
parser.add_argument("--account", required=True, help="confirmed Gitea account")
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
env = safe_environment()
|
|
result = subprocess.run(["git", "config", "--get", "remote.origin.url"], cwd=args.repo, env=env,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False)
|
|
if result.returncode:
|
|
raise ProbeError("Could not read the repository's origin URL.")
|
|
remote = result.stdout.decode("utf-8").strip()
|
|
git_read_url(args.site, remote)
|
|
helper = GitHelper(["git", "credential-manager"], env, os.path.abspath(args.repo))
|
|
token = load_credential(args.site, args.account, helper)
|
|
api, git = check_requests(args.site, args.account, token, remote, build_opener(NoRedirect()))
|
|
print("API authentication: " + api)
|
|
print("Git read access: " + git)
|
|
print("Read access does not prove permission to push.")
|
|
return 0 if api == "matched_account" and git == "read_advertisement" else 2
|
|
except (ProbeError, OSError, subprocess.SubprocessError, UnicodeError):
|
|
print("Credential check could not complete; no sensitive details are displayed.", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|