"""OAuth + synthetic activity/daily uploads to your own EffortSignal test account (Python 3.11+).""" import argparse import base64 import hashlib import json import secrets import time from datetime import datetime, time as day_time, timedelta, timezone from getpass import getpass from urllib.error import HTTPError from urllib.parse import parse_qs, urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener from zoneinfo import ZoneInfo class NoRedirect(HTTPRedirectHandler): # A redirect must not forward a bearer token or client secret to a different endpoint. def redirect_request(self, req, fp, code, msg, headers, newurl): return None def request_json(origin, path, *, method="GET", body=None, headers=None): request = Request(origin + path, data=body, headers=headers or {}, method=method) try: with build_opener(NoRedirect()).open(request, timeout=30) as response: return json.load(response) if response.status != 204 else None except HTTPError as error: # Responses can echo private inputs; expose only the status when a request fails. raise SystemExit(f"{method} {path.split('?')[0]}: HTTP {error.code}; see the guide's errors.") from None def token_request(origin, fields, client_id, secret): headers = {"Content-Type": "application/x-www-form-urlencoded"} if secret: # OAuth Basic encodes each credential before joining; secrets stay out of process arguments. credentials = f"{urlencode({'x': client_id})[2:]}:{urlencode({'x': secret})[2:]}" headers["Authorization"] = "Basic " + base64.b64encode(credentials.encode()).decode() return request_json(origin, "/oauth/token", method="POST", body=urlencode(fields).encode(), headers=headers) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--client-id", required=True) parser.add_argument("--redirect-uri", required=True) parser.add_argument("--public", action="store_true", help="Public client: PKCE, no secret") parser.add_argument("--disconnect", action="store_true", help="Revoke access after the example; history stays saved") parser.add_argument("--origin", default="https://app.effortsignal.com", help="Use the origin supplied for your registration") args = parser.parse_args(argv) origin = args.origin.rstrip("/") parsed_origin = urlsplit(origin) if (parsed_origin.scheme != "https" and not (parsed_origin.scheme == "http" and parsed_origin.hostname in {"127.0.0.1", "::1"})) or parsed_origin.path or parsed_origin.query or parsed_origin.fragment or parsed_origin.username or not parsed_origin.hostname: parser.error("Use an HTTPS origin (literal loopback HTTP is supported for isolated local QA).") resource = origin + "/api/v1/ingest" metadata = request_json(origin, "/.well-known/oauth-authorization-server?" + urlencode({"resource": resource})) if metadata.get("issuer") != origin or metadata.get("token_endpoint") != origin + "/oauth/token" or metadata.get("authorization_endpoint") != origin + "/oauth/authorize": raise SystemExit("Unexpected OAuth metadata; confirm the environment with EffortSignal.") secret = "" if args.public else getpass("Confidential client secret (hidden): ") if not args.public and not secret: raise SystemExit("The registered server client needs its own secret.") # Verifier/state exist only for this attempt; callback input and token responses are never logged. verifier = secrets.token_urlsafe(48) state = secrets.token_urlsafe(32) challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=") parameters = {"client_id": args.client_id, "redirect_uri": args.redirect_uri, "response_type": "code", "resource": resource, "scope": "activities:write sleep:write health:write", "state": state, "code_challenge": challenge, "code_challenge_method": "S256"} print("Use your own TEST athlete: this run saves a synthetic ride, sleep and resting pulse.") print("Open in your browser:\n" + origin + "/oauth/authorize?" + urlencode(parameters)) callback = urlsplit(getpass("Complete callback URL, within 60 seconds (hidden): ").strip()) registered = urlsplit(args.redirect_uri) query = parse_qs(callback.query) if (callback.scheme, callback.netloc, callback.path) != (registered.scheme, registered.netloc, registered.path) or callback.fragment: raise SystemExit("Callback address does not match this registration.") if query.get("state") != [state] or query.get("iss") != [metadata["issuer"]]: raise SystemExit("Callback state/issuer mismatch; no token exchange or upload was performed.") if "error" in query: raise SystemExit("Authorization was declined; no data was sent.") if len(query.get("code", [])) != 1: raise SystemExit("Callback needs one authorization code.") fields = {"grant_type": "authorization_code", "client_id": args.client_id, "redirect_uri": args.redirect_uri, "code": query["code"][0], "code_verifier": verifier, "resource": resource} tokens = token_request(origin, fields, args.client_id, secret) auth = {"Authorization": "Bearer " + tokens["access_token"]} connection = request_json(origin, "/api/v1/ingest/connection/", headers=auth) print("Connected; timezone:", connection["timezone"]) local_zone = ZoneInfo(connection["timezone"]) day = datetime.now(local_zone).date() - timedelta(days=1) source_clock = datetime.now(timezone.utc).isoformat() # Read the athlete's timezone instead of silently moving measurements to another local day. daily = {"timezone": connection["timezone"], "summaries": [{ "local_date": day.isoformat(), "source_updated_at": source_clock, "sleep_duration_seconds": 28800, "resting_heart_rate_bpm": 55, "metric_provenance": { "sleep_duration_seconds": {"device": "Synthetic example", "measurement_method": "main_sleep"}, "resting_heart_rate_bpm": {"device": "Synthetic example", "measurement_method": "provider_resting_heart_rate"}}}]} daily_result = request_json(origin, "/api/v1/ingest/daily-health/", method="POST", body=json.dumps(daily).encode(), headers={**auth, "Content-Type": "application/json"}) print("Daily batch: accepted", daily_result["accepted"], "; changed", daily_result["changed"]) external_id = "example-ride-" + secrets.token_hex(12) started = datetime.combine(day, day_time(9), tzinfo=local_zone) recording = {"record_type": "workout", "schema_version": 1, "workout_id": external_id, "source_fingerprint": hashlib.sha256(external_id.encode()).hexdigest(), "activity_type": "cycling", "sport": "cycling", "started_at": started.isoformat(), "ended_at": (started + timedelta(minutes=30)).isoformat(), "duration_seconds": 1800, "moving_time_seconds": 1800, "total_distance_meters": 8000, "source": {"device_name": "Synthetic example"}} recording_bytes = (json.dumps(recording) + "\n").encode() upload_path = "/api/v1/ingest/activities/?" + urlencode({"external_id": external_id, "source_updated_at": source_clock, "filename": external_id + ".eshealth"}) receipt = request_json(origin, upload_path, method="POST", body=recording_bytes, headers={**auth, "Content-Type": "application/octet-stream"}) print("Saved import receipt:", receipt["import_job_id"]) # A retry reuses exactly the accepted bytes/revision; it must not create a second import. replay = request_json(origin, upload_path, method="POST", body=recording_bytes, headers={**auth, "Content-Type": "application/octet-stream"}) if replay["import_job_id"] != receipt["import_job_id"]: raise SystemExit("Replay returned a different import receipt; contact EffortSignal.") tokens = token_request(origin, {"grant_type": "refresh_token", "client_id": args.client_id, "refresh_token": tokens["refresh_token"], "resource": resource}, args.client_id, secret) auth = {"Authorization": "Bearer " + tokens["access_token"]} receipt_path = "/api/v1/ingest/imports/" + receipt["import_job_id"] + "/" deadline = time.monotonic() + 60 while True: receipt = request_json(origin, receipt_path, headers=auth) if receipt["published"] or receipt["state"] in {"failed", "needs_review"} or receipt.get("error") or time.monotonic() >= deadline: break time.sleep(2) print("Import state:", receipt["state"], "; published:", receipt["published"]) if receipt.get("error") or receipt["state"] == "failed": print("Import needs attention; use the saved receipt ID when contacting support.") elif not receipt["published"]: print("Publication/review is still pending; acceptance is not publication.") if args.disconnect: request_json(origin, "/api/v1/ingest/connection/", method="DELETE", headers=auth) print("Connection revoked; accepted history remains saved.") print("Example finished. No secrets or tokens were saved to disk.") if __name__ == "__main__": main()