#!/usr/bin/env python3
"""requests-oauthlib 2.0.0 probe -- ONE python3 start, every measurement
99-python3-requests-oauthlib.test scores.

Emits `key=value' lines on stdout, one per line, no blank lines. The suite greps
each key out and compares it against a LITERAL measured on an x86-64 control
(CPython 3.12.3, requests-oauthlib 2.0.0 over oauthlib 3.2.2), so a red here is a
difference between this arm and that control rather than a difference from
something read off a README.

NO NETWORK, AND NO SERVER. Every case that would have made a request goes
through TapAdapter below, which requests mounts in place of HTTPAdapter: send()
never opens a socket -- it records the PreparedRequest and returns a committed
body. So the request the library built is asserted directly, which is both
stronger than asserting a server's answer and the only shape that is bounded and
deterministic on this arm.

THE PACKAGE IS AN ADAPTER, AND THAT IS WHY THE SPLIT OF CASES LOOKS AS IT DOES.
requests_oauthlib contributes no crypto and no protocol of its own: oauthlib
signs and parses, requests transports, and this package is the two auth handlers
(OAuth1/OAuth2) plus the two requests.Session subclasses that carry them. So the
measurements below are of the SEAM -- which header lands on the wire, which
exception reaches the caller, what the token dict looks like after a fetch --
and not of HMAC-SHA1, which is oauthlib's and is covered by BRINTOS-2406.

TWO RFC 5849 ANCHORS ARE ASSERTED BYTE-FOR-BYTE, and they are the one place the
expectations are not merely "what this version does". Section 3.4.1.1's worked
example prints its signature base string in full, so `rfc5849_base_string' below
is compared against the RFC's own bytes rather than against a control run.
"""
import base64
import hashlib
import hmac
import json
import os
import sys
import tempfile

OUT = []


def emit(key, value):
    """One `key=value' line, ALWAYS one line.

    Backslash, newline and carriage return are escaped: a raw newline would land
    as one counted result plus a stray line the runner cannot attribute, so the
    verdict would survive and the evidence would not."""
    text = str(value).replace("\\", "\\\\").replace("\n", "\\n").replace("\r", "\\r")
    OUT.append("%s=%s" % (key, text))


def yn(b):
    return "yes" if b else "no"


def attempt(fn):
    """`!ExceptionName' for a raise, else str(result).

    By exception NAME, not message: the class is what a caller codes against,
    and oauthlib's wording for several of these has moved between releases."""
    try:
        return str(fn())
    except BaseException as exc:          # noqa: BLE001 -- the class is the answer
        return "!" + type(exc).__name__


CWD_BEFORE = sorted(os.listdir("."))
ROOT = tempfile.mkdtemp(prefix="bst-reqoauth-")

# ---------------------------------------------------------------- identity ---
import requests                                                      # noqa: E402
import oauthlib                                                      # noqa: E402
import requests_oauthlib                                             # noqa: E402
from requests_oauthlib import OAuth1, OAuth1Session, OAuth2, OAuth2Session  # noqa: E402

emit("module_file", getattr(requests_oauthlib, "__file__", ""))
emit("is_package", yn(hasattr(requests_oauthlib, "__path__")))
emit("version_attr", getattr(requests_oauthlib, "__version__", "<absent>"))
try:
    import importlib.metadata as _md
    emit("version_metadata", _md.version("requests-oauthlib"))
except BaseException as exc:                                          # noqa: BLE001
    emit("version_metadata", "!" + type(exc).__name__)
emit("oauthlib_version", getattr(oauthlib, "__version__", "<absent>"))
emit("requests_version", getattr(requests, "__version__", "<absent>"))

# THE PACKAGE DECLARES NO __all__ -- measured, not assumed, because the obvious
# case to write here is `the names __all__ lists' and it would score an empty
# string against an empty string forever. What is asserted instead is that the
# four documented names are attributes of the package, which is the property a
# consumer's `from requests_oauthlib import ...' actually depends on.
emit("has_dunder_all", yn(hasattr(requests_oauthlib, "__all__")))
emit("exports_present", ",".join(
    n for n in ("OAuth1", "OAuth1Session", "OAuth2", "OAuth2Session")
    if hasattr(requests_oauthlib, n)))
emit("oauth1_is_authbase", yn(issubclass(OAuth1, requests.auth.AuthBase)))
emit("oauth2_is_authbase", yn(issubclass(OAuth2, requests.auth.AuthBase)))
emit("oauth1session_is_session", yn(issubclass(OAuth1Session, requests.Session)))
emit("oauth2session_is_session", yn(issubclass(OAuth2Session, requests.Session)))

# The package ships no compiled code: this image's python3 is ONE STATIC BINARY
# with an empty lib-dynload and no dynamic loader (BRINTOS-1476), so an
# extension could not be imported whatever it built to. Asserted from the
# imported module's own directory rather than from the recipe's `inherit'.
_pkgdir = os.path.dirname(requests_oauthlib.__file__)
emit("no_so_in_tree", yn(not [f for f in os.listdir(_pkgdir) if ".so" in f]))

# --------------------------------------------------------------- transport ---
# The whole of the no-network property lives in these thirty lines. send() is
# the ONLY path requests takes to a socket, and mounting this on "https://" and
# "http://" replaces it for every scheme the sessions below use.
class TapAdapter(requests.adapters.BaseAdapter):
    def __init__(self, status=200, body=b"", content_type="application/json"):
        super().__init__()
        self.status = status
        self.body = body
        self.content_type = content_type
        self.seen = []

    def send(self, request, **kwargs):
        self.seen.append(request)
        resp = requests.models.Response()
        resp.status_code = self.status
        resp._content = self.body
        resp.url = request.url
        resp.request = request
        resp.headers["Content-Type"] = self.content_type
        resp.encoding = "utf-8"
        return resp

    def close(self):
        pass


def mount(session, adapter):
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return adapter


# ===================================================================== OAuth1 ==
#
# RFC 5849 section 1.2's temporary-credentials request, with the nonce and the
# timestamp pinned so the header is a function of the inputs alone.
RFC_KEY = "dpf43f3p2l4k3l03"
RFC_SECRET = "kd94hf93k423kf44"

o1 = OAuth1Session(
    RFC_KEY,
    client_secret=RFC_SECRET,
    callback_uri="http://printer.example.com/ready",
    nonce="wIjqoS",
    timestamp="137131200",
    realm="Photos",
)
tap1 = mount(o1, TapAdapter(body=b"oauth_token=hh5s93j4hdidpola&oauth_token_secret=hdhd0244k9j7ao03&oauth_callback_confirmed=true",
                            content_type="application/x-www-form-urlencoded"))
r1 = o1.post("https://photos.example.net/initiate")

req1 = tap1.seen[0]
auth1 = req1.headers.get("Authorization")
if isinstance(auth1, bytes):
    auth1 = auth1.decode("ascii")
emit("oauth1_requests_made", len(tap1.seen))
emit("oauth1_method", req1.method)
emit("oauth1_url", req1.url)
emit("oauth1_header", auth1)

# The same header taken apart, so a red says WHICH parameter moved rather than
# only that the string differs.
_parts = dict(
    p.split("=", 1) for p in auth1[len("OAuth "):].split(", ")
)
emit("oauth1_param_names", ",".join(sorted(_parts)))
emit("oauth1_consumer_key", _parts.get("oauth_consumer_key", ""))
emit("oauth1_nonce", _parts.get("oauth_nonce", ""))
emit("oauth1_timestamp", _parts.get("oauth_timestamp", ""))
emit("oauth1_sigmethod", _parts.get("oauth_signature_method", ""))
emit("oauth1_version", _parts.get("oauth_version", ""))
emit("oauth1_realm", _parts.get("realm", ""))
emit("oauth1_callback", _parts.get("oauth_callback", ""))
emit("oauth1_signature", _parts.get("oauth_signature", ""))

# Nothing but the Authorization header was added, and the body stayed empty --
# a signature that had gone into the query string or the body would be a
# different wire request with the same header count.
emit("oauth1_body_empty", yn(not req1.body))
emit("oauth1_query_empty", yn("?" not in req1.url))

# Determinism: the same inputs a second time give the SAME signature, and moving
# the nonce alone moves it. Together these say the pin above is what makes the
# expectation above legitimate.
o1b = OAuth1Session(RFC_KEY, client_secret=RFC_SECRET,
                    callback_uri="http://printer.example.com/ready",
                    nonce="wIjqoS", timestamp="137131200", realm="Photos")
tapb = mount(o1b, TapAdapter())
o1b.post("https://photos.example.net/initiate")
_a = o1b.auth  # noqa: F841 -- kept for symmetry with the header read below
_hb = tapb.seen[0].headers.get("Authorization")
if isinstance(_hb, bytes):
    _hb = _hb.decode("ascii")
emit("oauth1_repeatable", yn(_hb == auth1))

o1c = OAuth1Session(RFC_KEY, client_secret=RFC_SECRET,
                    callback_uri="http://printer.example.com/ready",
                    nonce="OTHERNONCE", timestamp="137131200", realm="Photos")
tapc = mount(o1c, TapAdapter())
o1c.post("https://photos.example.net/initiate")
_hc = tapc.seen[0].headers.get("Authorization")
if isinstance(_hc, bytes):
    _hc = _hc.decode("ascii")
emit("oauth1_nonce_changes_signature", yn(_hc != auth1))

# THE RFC ANCHOR. RFC 5849 section 3.4.1.1 prints its signature base string in
# full; this is oauthlib's construction compared against those bytes, and it is
# the one expectation here that does not come from a control run.
from oauthlib.oauth1.rfc5849 import signature as _sig                # noqa: E402

_uri = ("http://example.com/request?b5=%3D%253D&a3=a&c%40=&a2=r%20b")
_body = "c2&a3=2+q"
_headers = {"Content-Type": "application/x-www-form-urlencoded",
            "Authorization": (
                'OAuth realm="Example", oauth_consumer_key="9djdj82h48djs9d2", '
                'oauth_token="kkk9d7dh3k39sjv7", oauth_signature_method="HMAC-SHA1", '
                'oauth_timestamp="137131201", oauth_nonce="7d8f3e4a", '
                'oauth_signature="djosJKDKJSD8743243%2Fjdk33klY%3D"')}
_params = _sig.collect_parameters(uri_query="b5=%3D%253D&a3=a&c%40=&a2=r%20b",
                                  body=_body, headers=_headers)
_norm = _sig.normalize_parameters(_params)
_bs = _sig.signature_base_string("POST", _sig.base_string_uri(_uri), _norm)
emit("rfc5849_base_string", _bs)
emit("rfc5849_base_string_uri", _sig.base_string_uri(_uri))

# ...and the HMAC-SHA1 of that base string, computed twice: once by oauthlib and
# once by hmac/hashlib straight out of the stdlib. Equal answers say the arm's
# hashlib and oauthlib agree; a red on `rfc5849_sig_matches_stdlib' separates a
# broken hash from a broken adapter, which no single number can.
_cs, _ts = "j49sk3j29djd", "dh893hdasih9"
_ol = _sig.sign_hmac_sha1(_bs, _cs, _ts)
_key = ("j49sk3j29djd" + "&" + "dh893hdasih9").encode("utf-8")
_manual = base64.b64encode(
    hmac.new(_key, _bs.encode("utf-8"), hashlib.sha1).digest()).decode("utf-8")
emit("rfc5849_signature", _ol)
emit("rfc5849_sig_matches_stdlib", yn(_ol == _manual))

# A SESSION WITH NO CREDENTIALS REFUSES TO SIGN. It does not fall through and
# send the request unsigned, which is the failure mode that would matter: the
# adapter below records how many requests reached it, and zero is the assertion.
o1none = OAuth1Session(None)
tapnone = mount(o1none, TapAdapter())
emit("oauth1_unsigned_refused",
     attempt(lambda: o1none.post("https://photos.example.net/initiate")))
emit("oauth1_unsigned_requests_made", len(tapnone.seen))

# The three-legged flow's parsers, over COMMITTED bodies -- no server answered
# either of these.
o1p = OAuth1Session(RFC_KEY, client_secret=RFC_SECRET)
tapp = mount(o1p, TapAdapter(
    body=b"oauth_token=hh5s93j4hdidpola&oauth_token_secret=hdhd0244k9j7ao03&oauth_callback_confirmed=true",
    content_type="application/x-www-form-urlencoded"))
_tok = o1p.fetch_request_token("https://photos.example.net/initiate")
emit("oauth1_request_token_keys", ",".join(sorted(_tok)))
emit("oauth1_request_token", _tok.get("oauth_token", ""))
emit("oauth1_request_token_secret", _tok.get("oauth_token_secret", ""))
emit("oauth1_callback_confirmed", _tok.get("oauth_callback_confirmed", ""))
emit("oauth1_token_is_dict", yn(isinstance(_tok, dict)))
emit("oauth1_session_token_set", yn(o1p._client.client.resource_owner_key == "hh5s93j4hdidpola"))

emit("oauth1_authorization_url",
     o1p.authorization_url("https://photos.example.net/authorize"))

# The callback URI parser is pure string work and is asserted on its own, since
# a red here is requests_oauthlib's and never oauthlib's transport.
_parsed = o1p.parse_authorization_response(
    "https://printer.example.com/ready?oauth_token=hh5s93j4hdidpola&oauth_verifier=hfdp7dh39dks9884")
emit("oauth1_parsed_keys", ",".join(sorted(_parsed)))
emit("oauth1_parsed_verifier", _parsed.get("oauth_verifier", ""))

# ===================================================================== OAuth2 ==
CLIENT_ID = "s6BhdRkqt3"
CLIENT_SECRET = "7Fjfp0ZBr1KtDRbnfVdmIw"
REDIRECT = "https://client.example.com/cb"
SCOPE = ["read", "write"]

o2 = OAuth2Session(client_id=CLIENT_ID, redirect_uri=REDIRECT, scope=SCOPE,
                   state="xyz")
_url, _state = o2.authorization_url("https://server.example.com/authorize")
emit("oauth2_authorization_url", _url)
emit("oauth2_state_returned", _state)
emit("oauth2_state_attr", o2.state)

# With no state given the session GENERATES one, and the generated value is what
# the URL carries. The value itself is random, so what is scored is the
# relationship and its length class, never the string.
o2gen = OAuth2Session(client_id=CLIENT_ID, redirect_uri=REDIRECT, scope=SCOPE)
_ugen, _sgen = o2gen.authorization_url("https://server.example.com/authorize")
emit("oauth2_state_generated_nonempty", yn(bool(_sgen)))
emit("oauth2_state_generated_in_url", yn(("state=" + _sgen) in _ugen))
emit("oauth2_state_generated_len_ge_16", yn(len(_sgen) >= 16))
emit("oauth2_state_generated_differs", yn(_sgen != _state))

emit("oauth2_authorized_before", yn(o2.authorized))

# --- token fetch, from a committed body ---------------------------------------
TOKEN_BODY = json.dumps({
    "access_token": "2YotnFZFEjr1zCsicMWpAA",
    "token_type": "Bearer",
    "expires_in": 3600,
    "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA",
    "scope": "read write",
}).encode("utf-8")

o2f = OAuth2Session(client_id=CLIENT_ID, redirect_uri=REDIRECT, scope=SCOPE,
                    state="xyz")
tapf = mount(o2f, TapAdapter(body=TOKEN_BODY))
_token = o2f.fetch_token(
    "https://server.example.com/token",
    client_secret=CLIENT_SECRET,
    authorization_response=REDIRECT + "?code=SplxlOBeZQQYbYS6WxSbIA&state=xyz",
)
emit("oauth2_token_keys", ",".join(sorted(_token)))
emit("oauth2_access_token", _token.get("access_token", ""))
emit("oauth2_token_type", _token.get("token_type", ""))
emit("oauth2_expires_in", _token.get("expires_in", ""))
emit("oauth2_refresh_token", _token.get("refresh_token", ""))
emit("oauth2_scope", ",".join(_token.get("scope", [])) if isinstance(_token.get("scope"), list) else _token.get("scope", ""))
emit("oauth2_expires_at_is_number", yn(isinstance(_token.get("expires_at"), float)))
emit("oauth2_authorized_after", yn(o2f.authorized))
emit("oauth2_access_token_attr", o2f.access_token)

# WHAT WENT OUT. The token request is where a client leaks its secret if the
# seam is wrong, so the captured request is asserted rather than assumed: the
# method, the target, the form encoding, the grant, the code, and that the
# secret travelled in the HTTP Basic header the RFC prefers.
_freq = tapf.seen[0]
_fbody = _freq.body.decode("utf-8") if isinstance(_freq.body, bytes) else (_freq.body or "")
emit("oauth2_token_requests_made", len(tapf.seen))
emit("oauth2_token_method", _freq.method)
emit("oauth2_token_url", _freq.url)
emit("oauth2_token_ctype", _freq.headers.get("Content-Type", ""))
emit("oauth2_token_body_params", ",".join(sorted(
    p.split("=", 1)[0] for p in _fbody.split("&") if p)))
emit("oauth2_token_body_grant", yn("grant_type=authorization_code" in _fbody))
emit("oauth2_token_body_code", yn("code=SplxlOBeZQQYbYS6WxSbIA" in _fbody))
_basic = _freq.headers.get("Authorization", "")
if isinstance(_basic, bytes):
    _basic = _basic.decode("ascii")
emit("oauth2_token_auth_scheme", _basic.split(" ", 1)[0] if _basic else "")
emit("oauth2_token_basic_decodes",
     base64.b64decode(_basic.split(" ", 1)[1]).decode("utf-8") if _basic.startswith("Basic ") else "")
emit("oauth2_secret_not_in_body", yn(CLIENT_SECRET not in _fbody))

# --- the signed request the token buys ----------------------------------------
o2use = OAuth2Session(client_id=CLIENT_ID, token=dict(json.loads(TOKEN_BODY)))
tapuse = mount(o2use, TapAdapter(body=b"{}"))
o2use.get("https://api.example.com/resource")
_ureq = tapuse.seen[0]
_uauth = _ureq.headers.get("Authorization", "")
if isinstance(_uauth, bytes):
    _uauth = _uauth.decode("ascii")
emit("oauth2_bearer_header", _uauth)

# --- refresh, from a second committed body ------------------------------------
REFRESH_BODY = json.dumps({
    "access_token": "NEWtnFZFEjr1zCsicMWpAA",
    "token_type": "Bearer",
    "expires_in": 3600,
    "refresh_token": "NEWv3JOkF0XG5Qx2TlKWIA",
    "scope": "read write",
}).encode("utf-8")

o2r = OAuth2Session(client_id=CLIENT_ID, token=dict(json.loads(TOKEN_BODY)))
tapr = mount(o2r, TapAdapter(body=REFRESH_BODY))
_new = o2r.refresh_token("https://server.example.com/token",
                         refresh_token="tGzv3JOkF0XG5Qx2TlKWIA",
                         client_id=CLIENT_ID, client_secret=CLIENT_SECRET)
emit("oauth2_refresh_keys", ",".join(sorted(_new)))
emit("oauth2_refresh_access_token", _new.get("access_token", ""))
emit("oauth2_refresh_new_refresh_token", _new.get("refresh_token", ""))
emit("oauth2_refresh_replaced_session_token",
     yn(o2r.token.get("access_token") == "NEWtnFZFEjr1zCsicMWpAA"))
_rreq = tapr.seen[0]
_rbody = _rreq.body.decode("utf-8") if isinstance(_rreq.body, bytes) else (_rreq.body or "")
emit("oauth2_refresh_body_grant", yn("grant_type=refresh_token" in _rbody))
emit("oauth2_refresh_requests_made", len(tapr.seen))

# --- the two documented error paths -------------------------------------------
# A state that does not match the one the session issued MUST raise before any
# token request is made -- so the adapter's request count is asserted beside the
# exception class, since raising after the exchange would leak the code.
o2m = OAuth2Session(client_id=CLIENT_ID, redirect_uri=REDIRECT, state="xyz")
tapm = mount(o2m, TapAdapter(body=TOKEN_BODY))
emit("oauth2_state_mismatch", attempt(lambda: o2m.fetch_token(
    "https://server.example.com/token", client_secret=CLIENT_SECRET,
    authorization_response=REDIRECT + "?code=SplxlOBeZQQYbYS6WxSbIA&state=WRONG")))
emit("oauth2_state_mismatch_requests_made", len(tapm.seen))

o2i = OAuth2Session(client_id=CLIENT_ID, redirect_uri=REDIRECT, state="xyz")
mount(o2i, TapAdapter(status=400, body=json.dumps(
    {"error": "invalid_grant", "error_description": "code expired"}).encode("utf-8")))
emit("oauth2_invalid_grant", attempt(lambda: o2i.fetch_token(
    "https://server.example.com/token", client_secret=CLIENT_SECRET,
    authorization_response=REDIRECT + "?code=SplxlOBeZQQYbYS6WxSbIA&state=xyz")))

o2e = OAuth2Session(client_id=CLIENT_ID, redirect_uri=REDIRECT, state="xyz")
mount(o2e, TapAdapter(status=400, body=json.dumps(
    {"error": "invalid_client"}).encode("utf-8")))
emit("oauth2_invalid_client", attempt(lambda: o2e.fetch_token(
    "https://server.example.com/token", client_secret=CLIENT_SECRET,
    authorization_response=REDIRECT + "?code=SplxlOBeZQQYbYS6WxSbIA&state=xyz")))

# An expired token with no auto-refresh wired up is the caller's to handle, and
# the class it arrives as is the contract.
_expired = dict(json.loads(TOKEN_BODY))
_expired["expires_at"] = 1.0
o2x = OAuth2Session(client_id=CLIENT_ID, token=_expired)
tapx = mount(o2x, TapAdapter(body=b"{}"))
emit("oauth2_expired", attempt(lambda: o2x.get("https://api.example.com/resource")))
emit("oauth2_expired_requests_made", len(tapx.seen))

# ...and with an auto-refresh URL wired up, the SAME expired token drives a
# refresh and then the resource request -- two requests through the tap, in that
# order, with no server involved in either.
class SeqAdapter(TapAdapter):
    def __init__(self, bodies):
        super().__init__()
        self.bodies = list(bodies)

    def send(self, request, **kwargs):
        self.body = self.bodies.pop(0) if self.bodies else b"{}"
        return super().send(request, **kwargs)


o2a = OAuth2Session(client_id=CLIENT_ID, token=dict(_expired),
                    auto_refresh_url="https://server.example.com/token",
                    auto_refresh_kwargs={"client_id": CLIENT_ID,
                                         "client_secret": CLIENT_SECRET},
                    token_updater=lambda t: None)
tapa = mount(o2a, SeqAdapter([REFRESH_BODY, b"{}"]))
emit("oauth2_autorefresh", attempt(lambda: o2a.get("https://api.example.com/resource")))
emit("oauth2_autorefresh_requests_made", len(tapa.seen))
emit("oauth2_autorefresh_first_url", tapa.seen[0].url if tapa.seen else "")
emit("oauth2_autorefresh_token_replaced",
     yn(o2a.token.get("access_token") == "NEWtnFZFEjr1zCsicMWpAA"))

# A token_updater is REQUIRED once auto-refresh is armed; leaving it out is the
# documented TokenUpdated exception rather than a silent drop of the new token.
o2u = OAuth2Session(client_id=CLIENT_ID, token=dict(_expired),
                    auto_refresh_url="https://server.example.com/token",
                    auto_refresh_kwargs={"client_id": CLIENT_ID,
                                         "client_secret": CLIENT_SECRET})
mount(o2u, SeqAdapter([REFRESH_BODY, b"{}"]))
emit("oauth2_no_updater", attempt(lambda: o2u.get("https://api.example.com/resource")))

# --- compliance hooks ----------------------------------------------------------
# The four hook names are the package's whole extension surface for servers that
# do not answer to spec, so the set is asserted by name.
emit("oauth2_compliance_hooks", ",".join(sorted(o2.compliance_hook)))

# ------------------------------------------------------------------ hygiene ---
# NOTHING WAS WRITTEN OUTSIDE THE WORKDIR, and no socket was opened: every
# session above was mounted on a tap, so a non-zero count here would mean a
# code path found a transport this probe did not replace.
os.rmdir(ROOT)
emit("workdir_removed", yn(not os.path.exists(ROOT)))
emit("cwd_unchanged", yn(sorted(os.listdir(".")) == CWD_BEFORE))

emit("emitted_count", len(OUT) + 2)
emit("probe_done", "yes")
sys.stdout.write("\n".join(OUT) + "\n")