taler-devtesting-exchange-challenger-setup (11172B)
1 #!/usr/bin/env python3 2 """ 3 Taler Challenger Address Verification Setup Tool 4 ================================================= 5 This script reads the configuration for a KYC provider section from a taler-exchange config, 6 sends a setup request to the Challenger API, and returns the address verification URL. 7 """ 8 9 import argparse 10 import json 11 import logging 12 import secrets 13 import shutil 14 import subprocess 15 import sys 16 import urllib.error 17 import urllib.parse 18 import urllib.request 19 20 21 def list_sections(config_file): 22 """List available configuration sections using taler-exchange-config.""" 23 cmd = ["taler-exchange-config"] 24 if config_file: 25 cmd.extend(["-c", config_file]) 26 cmd.append("-S") 27 try: 28 res = subprocess.run(cmd, capture_output=True, text=True, check=True) 29 lines = res.stdout.strip().split("\n") 30 sections = [] 31 for line in lines: 32 line = line.strip() 33 if line and not line.startswith("The following sections are available"): 34 sections.append(line) 35 return sections 36 except subprocess.CalledProcessError as e: 37 print(f"Warning: failed to list sections: {e.stderr.strip()}", file=sys.stderr) 38 return [] 39 40 41 def get_config_val(config_file, section, option): 42 """Retrieve a configuration value using taler-exchange-config.""" 43 cmd = ["taler-exchange-config"] 44 if config_file: 45 cmd.extend(["-c", config_file]) 46 cmd.extend(["-s", section, "-o", option]) 47 res = subprocess.run(cmd, capture_output=True, text=True) 48 if res.returncode == 0: 49 return res.stdout.strip() 50 return None 51 52 53 def resolve_section(config_file, section_arg): 54 """Resolve the section name by checking exact matches and kyc-provider prefixes/suffixes.""" 55 sections = list_sections(config_file) 56 if not sections: 57 return section_arg 58 59 # Try exact match 60 if section_arg in sections: 61 return section_arg 62 63 # Try prepending 'kyc-provider-' 64 prefixed = f"kyc-provider-{section_arg}" 65 if prefixed in sections: 66 return prefixed 67 68 # Try suffixing with '-kyc-provider' 69 suffixed = f"{section_arg}-kyc-provider" 70 if suffixed in sections: 71 return suffixed 72 73 # Try finding any section containing section_arg 74 matches = [s for s in sections if section_arg in s] 75 if len(matches) == 1: 76 return matches[0] 77 elif len(matches) > 1: 78 print(f"Warning: Ambiguous section name '{section_arg}'. Matches: {', '.join(matches)}", file=sys.stderr) 79 80 return section_arg 81 82 83 def main(): 84 if not shutil.which("taler-exchange-config"): 85 print("Error: 'taler-exchange-config' command not found in PATH.", file=sys.stderr) 86 sys.exit(1) 87 88 parser = argparse.ArgumentParser( 89 description="Initiate a Challenger address verification process using taler-exchange configuration." 90 ) 91 parser.add_argument( 92 "section", 93 help="The configuration section name (with or without 'kyc-provider-' prefix/suffix, e.g., 'postal-challenger')" 94 ) 95 parser.add_argument( 96 "-c", "--config", 97 help="Path to the taler-exchange configuration file (e.g., challenger-test.conf)" 98 ) 99 parser.add_argument( 100 "-r", "--redirect-uri", 101 help="OAuth2 redirect URI (falls back to exchange BASE_URL/kyc-proof/<provider_name>)" 102 ) 103 parser.add_argument( 104 "-s", "--state", 105 help="OAuth2 state parameter (default: cryptographically secure random string)" 106 ) 107 parser.add_argument( 108 "-o", "--read-only", 109 action="store_true", 110 help="Instruct Challenger to prevent editing of the pre-populated address" 111 ) 112 parser.add_argument( 113 "--json", 114 action="store_true", 115 help="Output results in JSON format" 116 ) 117 parser.add_argument( 118 "-v", "--verbose", 119 action="store_true", 120 help="Log verbose HTTP request/response details" 121 ) 122 parser.add_argument( 123 "--address-json", 124 help="Raw JSON object containing fields to pre-populate the address" 125 ) 126 127 # Individual address fields for convenience 128 parser.add_argument("--email", help="Pre-populate CONTACT_EMAIL (for email validation)") 129 parser.add_argument("--phone", help="Pre-populate CONTACT_PHONE (for phone validation)") 130 parser.add_argument("--name", help="Pre-populate CONTACT_NAME (for postal validation)") 131 parser.add_argument("--address-lines", help="Pre-populate ADDRESS_LINES (for postal validation)") 132 parser.add_argument("--country", help="Pre-populate ADDRESS_COUNTRY (for postal validation)") 133 134 args = parser.parse_args() 135 136 # Configure logging 137 log_level = logging.DEBUG if args.verbose else logging.INFO 138 logging.basicConfig( 139 level=log_level, 140 format="%(asctime)s [%(levelname)s] %(message)s", 141 stream=sys.stderr 142 ) 143 144 # Resolve section name 145 section = resolve_section(args.config, args.section) 146 147 # Read required OAuth2 configurations 148 client_id = get_config_val(args.config, section, "KYC_OAUTH2_CLIENT_ID") 149 client_secret = get_config_val(args.config, section, "KYC_OAUTH2_CLIENT_SECRET") 150 authorize_url = get_config_val(args.config, section, "KYC_OAUTH2_AUTHORIZE_URL") 151 152 missing = [] 153 if not client_id: 154 missing.append("KYC_OAUTH2_CLIENT_ID") 155 if not client_secret: 156 missing.append("KYC_OAUTH2_CLIENT_SECRET") 157 if not authorize_url: 158 missing.append("KYC_OAUTH2_AUTHORIZE_URL") 159 160 if missing: 161 print(f"Error: Missing required options in config section [{section}]: {', '.join(missing)}", file=sys.stderr) 162 sys.exit(1) 163 164 # Construct the Setup URL 165 # E.g. https://postal.challenger.stage.taler-ops.ch/authorize#setup -> https://postal.challenger.stage.taler-ops.ch/setup/1 166 url_no_frag = authorize_url.split('#')[0] 167 if '/' not in url_no_frag: 168 print(f"Error: Invalid KYC_OAUTH2_AUTHORIZE_URL: {authorize_url}", file=sys.stderr) 169 sys.exit(1) 170 base_part = url_no_frag.rsplit('/', 1)[0] 171 setup_url = f"{base_part}/setup/{client_id}" 172 173 # Determine redirect URI 174 redirect_uri = args.redirect_uri 175 if not redirect_uri: 176 exchange_base_url = get_config_val(args.config, "exchange", "BASE_URL") 177 if exchange_base_url: 178 provider_name = section 179 if provider_name.startswith("kyc-provider-"): 180 provider_name = provider_name[len("kyc-provider-"):] 181 if not exchange_base_url.endswith("/"): 182 exchange_base_url += "/" 183 redirect_uri = f"{exchange_base_url}kyc-proof/{provider_name}" 184 185 # Determine state parameter 186 state = args.state if args.state else secrets.token_hex(16) 187 188 # Build Challenger setup request payload 189 payload = {"read_only": args.read_only} 190 191 if args.address_json: 192 try: 193 json_data = json.loads(args.address_json) 194 payload.update(json_data) 195 except json.JSONDecodeError as e: 196 print(f"Error: Invalid JSON payload in --address-json: {e}", file=sys.stderr) 197 sys.exit(1) 198 199 if args.email: 200 payload["CONTACT_EMAIL"] = args.email 201 if args.phone: 202 payload["CONTACT_PHONE"] = args.phone 203 if args.name: 204 payload["CONTACT_NAME"] = args.name 205 if args.address_lines: 206 payload["ADDRESS_LINES"] = args.address_lines 207 if args.country: 208 payload["ADDRESS_COUNTRY"] = args.country 209 210 # Optional scope configuration 211 scope = get_config_val(args.config, section, "KYC_OAUTH2_SCOPE") 212 213 # Make the HTTP setup POST request to Challenger 214 req_data = json.dumps(payload).encode("utf-8") 215 req = urllib.request.Request( 216 setup_url, 217 data=req_data, 218 headers={ 219 "Authorization": f"Bearer {client_secret}", 220 "Content-Type": "application/json" 221 }, 222 method="POST" 223 ) 224 225 logging.info(f"Making POST request to {setup_url}") 226 # Redact client secret in headers for safety 227 headers_logged = dict(req.headers) 228 if 'Authorization' in headers_logged: 229 headers_logged['Authorization'] = 'Bearer <redacted>' 230 logging.debug(f"Request Headers: {headers_logged}") 231 logging.debug(f"Request Body: {payload}") 232 233 try: 234 with urllib.request.urlopen(req) as response: 235 resp_bytes = response.read() 236 logging.info(f"Received HTTP {response.status} from Challenger") 237 logging.debug(f"Response Headers: {dict(response.getheaders())}") 238 logging.debug(f"Response Body: {resp_bytes.decode('utf-8')}") 239 resp_json = json.loads(resp_bytes.decode("utf-8")) 240 except urllib.error.HTTPError as e: 241 err_bytes = e.read() 242 err_msg = err_bytes.decode("utf-8") 243 logging.error(f"HTTP setup request failed with status code {e.code}: {e.reason}") 244 logging.debug(f"Response Headers: {dict(e.headers)}") 245 logging.debug(f"Response Body: {err_msg}") 246 247 print(f"Error: HTTP setup request failed with status code {e.code}: {e.reason}", file=sys.stderr) 248 try: 249 err_json = json.loads(err_msg) 250 print(json.dumps(err_json, indent=2), file=sys.stderr) 251 except json.JSONDecodeError: 252 if err_msg: 253 print(err_msg, file=sys.stderr) 254 sys.exit(1) 255 except urllib.error.URLError as e: 256 logging.error(f"Network connection failed: {e.reason}") 257 print(f"Error: Network connection failed to {setup_url}: {e.reason}", file=sys.stderr) 258 sys.exit(1) 259 260 nonce = resp_json.get("nonce") 261 if not nonce: 262 print("Error: Challenger response did not contain a 'nonce'.", file=sys.stderr) 263 print(json.dumps(resp_json, indent=2), file=sys.stderr) 264 sys.exit(1) 265 266 # Build redirect / authorization URL 267 auth_base = f"{url_no_frag}/{nonce}" if not url_no_frag.endswith("/") else f"{url_no_frag}{nonce}" 268 query_params = { 269 "response_type": "code", 270 "client_id": client_id, 271 "state": state 272 } 273 if redirect_uri: 274 query_params["redirect_uri"] = redirect_uri 275 if scope: 276 query_params["scope"] = scope 277 278 encoded_params = urllib.parse.urlencode(query_params) 279 auth_url = f"{auth_base}?{encoded_params}" 280 281 if args.json: 282 out = { 283 "setup_url": setup_url, 284 "nonce": nonce, 285 "client_id": client_id, 286 "redirect_uri": redirect_uri, 287 "state": state, 288 "scope": scope, 289 "authorization_url": auth_url 290 } 291 print(json.dumps(out, indent=2)) 292 else: 293 print("Successfully initiated address verification with Challenger!") 294 print(f"Challenger Setup URL: {setup_url}") 295 print(f"Returned Nonce: {nonce}") 296 print(f"Client ID: {client_id}") 297 print(f"Redirect URI: {redirect_uri if redirect_uri else 'Not specified'}") 298 print(f"State: {state}") 299 if scope: 300 print(f"Scope: {scope}") 301 print("\nTo complete address verification, open the following URL in your browser:") 302 print(auth_url) 303 304 305 if __name__ == "__main__": 306 main()