"""ShopSafe REST example. Python 3.11+, pip install httpx. Set SHOPSAFE_API_KEY in your secret store/environment, then run: python shopsafe_verify.py https://example.com --request-id order-123-url-check-1 The default maximum is zero. Paid work requires the owner's explicit authority: ... --max-cost-cents 2 Persist the request ID AND identical URL/cost/age parameters with your job before the first call. Reuse them after a timeout or process restart. Never generate a new ID merely to retry. No automatic funding, cap changes or transaction actions. """ from __future__ import annotations import argparse import json import os import re import sys import time from datetime import datetime, timezone import httpx BASE_URL = 'https://shopsafe.com.au' class ShopSafeError(Exception): def __init__(self, detail): self.detail = detail super().__init__(str(detail.get('error', 'request_unavailable'))) def _uncertain(code='transport_interrupted'): return ShopSafeError({'error':code, 'retryable':True, 'next_action':'wait_and_retry', 'retry_after_seconds':1, 'charge_status':'check_original_request', 'retry_instruction':'Retry with the original request ID and identical parameters. Do not assume that no charge occurred.'}) class ShopSafe: def __init__(self, api_key, *, transport=None): if not api_key or not isinstance(api_key,str) or any(ord(c)<33 or ord(c)>126 for c in api_key): raise ValueError('Supply the API key through your secret store.') self.client = httpx.Client(headers={'Authorization':'Bearer '+api_key,'Accept':'application/json'}, timeout=httpx.Timeout(45,connect=5),follow_redirects=False,transport=transport) def close(self): self.client.close() def __enter__(self): return self def __exit__(self,*_): self.close() def _request(self,method,path,payload=None): try: response = self.client.request(method,BASE_URL+path,json=payload) except httpx.TransportError: raise _uncertain() from None # Never follow a redirect with the service credential or expose raw # gateway HTML, request headers or internal transport exception text. if 300 <= response.status_code < 400: raise ShopSafeError({'error':'unexpected_redirect','retryable':False, 'next_action':'contact_support','charge_status':'check_original_request'}) try: data = response.json() except ValueError: raise _uncertain('invalid_service_response') from None if not isinstance(data,dict): raise _uncertain('invalid_service_response') if response.status_code != 200: if not isinstance(data.get('error'),str) or not isinstance(data.get('next_action'),str): raise _uncertain('invalid_service_response') raise ShopSafeError(data) return data def account(self): return self._request('GET','/api/v1/account') def verification(self,verification_id): if not isinstance(verification_id,str) or not re.fullmatch(r'[a-zA-Z0-9_-]{1,64}',verification_id): raise ValueError('Use the verification ID returned by the service.') return self._request('GET','/api/v1/verifications/'+verification_id) def verify(self,url,request_id,*,max_cost_cents=0,max_age_seconds=300): if not isinstance(request_id,str) or not re.fullmatch(r'[a-zA-Z0-9._-]{1,64}',request_id): raise ValueError('Supply a persistent request ID of 1 to 64 letters, digits, dots, underscores or hyphens.') if type(max_cost_cents) is not int or not 0 <= max_cost_cents <= 100: raise ValueError('The authorised maximum must be an integer number of AUD cents from 0 to 100.') if type(max_age_seconds) is not int or not 0 <= max_age_seconds <= 300: raise ValueError('The maximum evidence age must be 0 to 300 seconds.') if not isinstance(url,str) or not 1 <= len(url) <= 2048: raise ValueError('Supply the public URL to check, without credentials or login/reset secrets.') return self._request('POST','/api/v1/verifications',{'url':url,'idempotency_key':request_id, 'max_cost_aud_cents':max_cost_cents,'max_age_seconds':max_age_seconds}) def current_complete_evidence(result,*,now=None): """Freshness/coverage helper only. True does not authorise a transaction.""" if not isinstance(result,dict): return False verdict = result.get('verdict') or {} if not isinstance(verdict,dict): return False try: expires = datetime.fromisoformat(verdict['expires_at'].replace('Z','+00:00')) return bool(result.get('status')=='complete' and result.get('freshness')=='fresh' and verdict.get('coverage')=='complete' and expires.tzinfo is not None and expires > (now or datetime.now(timezone.utc))) except (ValueError,KeyError,TypeError,AttributeError): return False def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('url') parser.add_argument('--request-id',required=True) parser.add_argument('--max-cost-cents',type=int,default=0) args = parser.parse_args() try: with ShopSafe(os.environ.get('SHOPSAFE_API_KEY','')) as api: # Inspect pricing and owner limits; never change them automatically. account = api.account() if args.max_cost_cents and account.get('quick_price_cents',101) > args.max_cost_cents: raise ShopSafeError({'error':'owner_authorisation_required','retryable':False, 'next_action':'owner_authorisation','charge_status':'not_charged'}) for attempt in range(3): try: result = api.verify(args.url,args.request_id,max_cost_cents=args.max_cost_cents) print(json.dumps(result)) return 0 if result.get('status')=='complete' else 3 except ShopSafeError as error: wait = error.detail.get('retry_after_seconds') # Only short retries. Longer waits and payment decisions go # back to the calling agent/owner with the original details. if error.detail.get('retryable') is not True or type(wait) is not int or not 0 <= wait <= 2 or attempt==2: raise time.sleep(max(1,wait)) except ShopSafeError as error: print(json.dumps(error.detail)) return 2 except ValueError: print(json.dumps({'error':'invalid_client_configuration','next_action':'correct_request'})) return 2 if __name__=='__main__': sys.exit(main())