sashatrask
213 lines
8.0 KiB
Python
213 lines
8.0 KiB
Python
import sys
|
|
|
|
sys.dont_write_bytecode = True
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
|
|
import requests
|
|
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from keycheck_common import (
|
|
append_jsonl,
|
|
commit_status_transaction,
|
|
default_input_file,
|
|
default_proxy_file,
|
|
ensure_output_files,
|
|
iter_findings,
|
|
load_checked_statuses,
|
|
load_known_keys,
|
|
load_proxies,
|
|
mask_secret,
|
|
read_plain_keys,
|
|
recover_status_transaction,
|
|
request_error_message,
|
|
record_validation_result,
|
|
require_provider_authority,
|
|
service_output_dir,
|
|
should_skip_key,
|
|
write_keycheck_event,
|
|
)
|
|
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
PARENT_DIR = os.path.dirname(SCRIPT_DIR)
|
|
SERVICE = "github"
|
|
OUTPUT_DIR = os.getenv("KEYCHECK_OUTPUT_DIR") or service_output_dir(SERVICE)
|
|
|
|
INPUT_FILE = os.getenv("KEYCHECK_INPUT_FILE") or default_input_file()
|
|
PROXY_FILE = os.getenv("KEYCHECK_PROXY_FILE") or default_proxy_file()
|
|
PLAIN_FILE = os.path.join(OUTPUT_DIR, "github.txt")
|
|
CHECKED_FILE = os.path.join(OUTPUT_DIR, "githubChecked.txt")
|
|
RESULTS_FILE = os.path.join(OUTPUT_DIR, "githubResults.jsonl")
|
|
|
|
STATUS_FILES = {
|
|
"VALID": os.path.join(OUTPUT_DIR, "githubAlive.txt"),
|
|
"DEAD": os.path.join(OUTPUT_DIR, "githubDead.txt"),
|
|
"RESTRICTED": os.path.join(OUTPUT_DIR, "githubRestricted.txt"),
|
|
"RATE_LIMITED": os.path.join(OUTPUT_DIR, "githubRateLimited.txt"),
|
|
"NETWORK": os.path.join(OUTPUT_DIR, "githubNetwork.txt"),
|
|
"UNKNOWN": os.path.join(OUTPUT_DIR, "githubUnknown.txt"),
|
|
"REFRESH_TOKEN": os.path.join(OUTPUT_DIR, "githubRefreshToken.txt"),
|
|
}
|
|
|
|
GITHUB_TOKEN_RE = re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b")
|
|
|
|
|
|
def ensure_files():
|
|
ensure_output_files([CHECKED_FILE, RESULTS_FILE, *STATUS_FILES.values(), PLAIN_FILE])
|
|
recover_status_transaction(CHECKED_FILE, STATUS_FILES)
|
|
|
|
|
|
def extract_candidates(input_file, plain_files):
|
|
seen_plain = set()
|
|
for item in iter_findings(input_file, ["Github", "GitHubOauth2"]):
|
|
text = "\n".join(str(value or "") for value in [item.get("raw"), item.get("raw_v2")])
|
|
for match in GITHUB_TOKEN_RE.findall(text):
|
|
yield match, item["source"], item["finding"]
|
|
|
|
for item in read_plain_keys(plain_files, GITHUB_TOKEN_RE):
|
|
key = item["key"]
|
|
if key not in seen_plain:
|
|
seen_plain.add(key)
|
|
yield key, item["source"], {}
|
|
|
|
|
|
def is_rate_limited(response):
|
|
remaining = response.headers.get("X-RateLimit-Remaining")
|
|
return response.status_code in (403, 429) and remaining == "0"
|
|
|
|
|
|
def check_token(token, proxy, timeout):
|
|
if token.startswith("ghr_"):
|
|
return {
|
|
"status": "REFRESH_TOKEN",
|
|
"message": "GitHub refresh tokens cannot be checked directly as bearer API tokens",
|
|
}
|
|
|
|
if token.startswith("ghs_"):
|
|
url = "https://api.github.com/installation/repositories"
|
|
token_kind = "installation"
|
|
else:
|
|
url = "https://api.github.com/user"
|
|
token_kind = "user"
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/vnd.github+json",
|
|
"X-GitHub-Api-Version": "2022-11-28",
|
|
"User-Agent": "local-keycheck-github",
|
|
}
|
|
try:
|
|
response = requests.get(url, headers=headers, proxies=proxy, timeout=timeout)
|
|
except requests.RequestException as exc:
|
|
return {"status": "NETWORK", "message": str(exc), "token_kind": token_kind}
|
|
|
|
message = request_error_message(response)
|
|
scopes = response.headers.get("X-OAuth-Scopes", "")
|
|
accepted_scopes = response.headers.get("X-Accepted-OAuth-Scopes", "")
|
|
rate_remaining = response.headers.get("X-RateLimit-Remaining", "")
|
|
rate_reset = response.headers.get("X-RateLimit-Reset", "")
|
|
|
|
if response.status_code == 200:
|
|
payload = response.json()
|
|
extra = {
|
|
"token_kind": token_kind,
|
|
"scopes": scopes,
|
|
"accepted_scopes": accepted_scopes,
|
|
"rate_remaining": rate_remaining,
|
|
"rate_reset": rate_reset,
|
|
}
|
|
if token_kind == "installation":
|
|
extra["repo_count"] = payload.get("total_count")
|
|
return {"status": "VALID", "message": "installation token accepted", **extra}
|
|
return {
|
|
"status": "VALID",
|
|
"message": "user token accepted",
|
|
"login": payload.get("login"),
|
|
"account_type": payload.get("type"),
|
|
**extra,
|
|
}
|
|
|
|
if response.status_code == 401:
|
|
return {"status": "DEAD", "http_status": 401, "message": message, "token_kind": token_kind}
|
|
if is_rate_limited(response):
|
|
return {"status": "RATE_LIMITED", "http_status": response.status_code, "message": message, "token_kind": token_kind, "rate_reset": rate_reset}
|
|
if response.status_code == 403:
|
|
return {"status": "RESTRICTED", "http_status": 403, "message": message, "token_kind": token_kind, "scopes": scopes}
|
|
if response.status_code in (404, 422):
|
|
return {"status": "UNKNOWN", "http_status": response.status_code, "message": message, "token_kind": token_kind}
|
|
if response.status_code >= 500:
|
|
return {"status": "NETWORK", "http_status": response.status_code, "message": message, "token_kind": token_kind}
|
|
return {"status": "UNKNOWN", "http_status": response.status_code, "message": message, "token_kind": token_kind}
|
|
|
|
|
|
def write_result(key, result, source, finding):
|
|
write_keycheck_event(SERVICE, RESULTS_FILE, key, result, source, finding)
|
|
extra = result.get("login") or result.get("repo_count") or result.get("token_kind") or source
|
|
commit_status_transaction(
|
|
CHECKED_FILE, STATUS_FILES, key, result["status"], result.get("message", ""), extra,
|
|
)
|
|
record_validation_result(SERVICE, key, result, source, finding)
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="GitHub token checker")
|
|
parser.add_argument("--input", default=INPUT_FILE)
|
|
parser.add_argument("--plain", action="append", default=[])
|
|
parser.add_argument("--proxy-file", default=PROXY_FILE)
|
|
parser.add_argument("--timeout", type=int, default=20)
|
|
parser.add_argument("--max-keys", type=int, default=0)
|
|
parser.add_argument("--retry-network", action="store_true")
|
|
parser.add_argument("--retry-limited", action="store_true")
|
|
parser.add_argument("--retry-unknown", action="store_true")
|
|
parser.add_argument("--retry-restricted", action="store_true")
|
|
parser.add_argument("--recheck-all", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main():
|
|
require_provider_authority(SERVICE)
|
|
args = parse_args()
|
|
ensure_files()
|
|
proxy_cycler = load_proxies(args.proxy_file)
|
|
checked = load_checked_statuses(CHECKED_FILE)
|
|
known = load_known_keys(CHECKED_FILE, STATUS_FILES)
|
|
retry_statuses = set()
|
|
if args.retry_network:
|
|
retry_statuses.add("NETWORK")
|
|
if args.retry_limited:
|
|
retry_statuses.add("RATE_LIMITED")
|
|
if args.retry_unknown:
|
|
retry_statuses.add("UNKNOWN")
|
|
if args.retry_restricted:
|
|
retry_statuses.add("RESTRICTED")
|
|
|
|
plain_files = args.plain or [PLAIN_FILE]
|
|
processed = 0
|
|
skipped = 0
|
|
for key, source, finding in extract_candidates(args.input, plain_files):
|
|
if should_skip_key(key, checked, known, args, retry_statuses, service=SERVICE, source=source, finding=finding):
|
|
skipped += 1
|
|
continue
|
|
if args.max_keys and processed >= args.max_keys:
|
|
break
|
|
processed += 1
|
|
print(f"\n[{processed}] GitHub candidate {mask_secret(key)} from {source}")
|
|
proxy = next(proxy_cycler) if proxy_cycler else None
|
|
result = check_token(key, proxy, args.timeout)
|
|
print(f" STATUS: {result['status']} | {str(result.get('message', ''))[:200]}")
|
|
write_result(key, result, source, finding)
|
|
known.add(key)
|
|
checked[key] = result["status"]
|
|
time.sleep(0.1)
|
|
print(f"\nDone. Processed={processed}, skipped={skipped}, results={RESULTS_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|