"""InFinBot standalone paper account and fresh crypto REST quote diagnostic. Python 3.8+ standard library only; no companion modules or pip packages. GET requests only. No orders, ledger writes, mock credentials or fallback prices. Futures and streaming remain unverified. Authentication cannot prove rotation. """ import argparse import copy import json import os import re import shlex import stat from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit, parse_qs UTC = timezone.utc from infinbot_proxy import PROXY_BASE_URL,read_proxy_json,ProxyError PAPER = PROXY_BASE_URL DATA = PROXY_BASE_URL ALLOWED = frozenset({'ALPACA_BASE_URL', 'ALPACA_PAPER_KEY', 'ALPACA_PAPER_SECRET', 'ALPACA_LIVE_KEY', 'ALPACA_LIVE_SECRET'}) class Rejected(ValueError): pass def number(value, positive=False): try: result = Decimal(str(value)) except (InvalidOperation, ValueError): raise Rejected('Invalid decimal') from None if not result.is_finite() or result < 0 or (positive and result <= 0): raise Rejected('Expected finite nonnegative number') return result def instant(value): try: result = datetime.fromisoformat(str(value).replace('Z', '+00:00')) except ValueError: raise Rejected('Invalid timestamp') from None if result.tzinfo is None: raise Rejected('Timestamp must include timezone') return result.astimezone(UTC) class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise Rejected('Redirect blocked') def account_amount(value): try:value=Decimal(str(value)) except (InvalidOperation,ValueError):raise Rejected('Malformed account amount') from None if not value.is_finite():raise Rejected('Nonfinite account amount') return str(value) def load_private_credentials(path): """Read only literal, allowlisted assignments from an owner-private POSIX file. No shell execution, environment mutation, interpolation, or credential values in error messages. Symlinks, shared files and oversized files fail. """ if not all(hasattr(os, name) for name in ('O_NOFOLLOW', 'geteuid')): raise Rejected('Private file loading requires supported POSIX ownership controls') try: fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | getattr(os, 'O_NONBLOCK', 0)) with os.fdopen(fd, 'rb') as stream: info = os.fstat(stream.fileno()) if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077 or info.st_uid != os.geteuid(): raise Rejected('Credential file must be owner-private and owned by the current user') if info.st_nlink != 1: raise Rejected('Credential file must not have multiple hard links') raw = stream.read(8193) if len(raw) > 8192: raise Rejected('Credential file exceeds the size limit') text = raw.decode('utf-8') except (OSError, UnicodeError): raise Rejected('Private credential file unavailable or unsafe') from None result = {} for number, line in enumerate(text.splitlines(), 1): line = line.strip() if not line or line.startswith('#'): continue if line.startswith('export '): line = line[7:].strip() if '=' not in line: raise Rejected(f'Invalid credential assignment on line {number}') key, value = (part.strip() for part in line.split('=', 1)) if key not in ALLOWED or key in result: raise Rejected(f'Unsupported or duplicate credential assignment on line {number}') if any(c in value for c in ('$', '`', '\\', '\x00')): raise Rejected(f'Literal credential value required on line {number}') try: tokens = shlex.split(value, comments=True, posix=True) except ValueError: raise Rejected(f'Invalid literal credential assignment on line {number}') from None if len(tokens) != 1 or any(c in tokens[0] for c in ('$','`','\x00','\r','\n','\\')): raise Rejected(f'Literal credential value required on line {number}') result[key] = tokens[0] return result def account_credentials(settings, mode): # Proxy infrastructure owns broker authentication. Never inspect client keys. if mode != 'paper':raise Rejected('Declared paper diagnostics only; proxy upstream mode remains unverified') return PROXY_BASE_URL, {} def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise Rejected('Duplicate configuration key') result[key] = value return result def load_worker_config(path): with open(path, 'rb') as stream: raw = stream.read(16385) if len(raw) > 16384: raise Rejected('Configuration too large') config = json.loads(raw, object_pairs_hook=unique_object, parse_constant=reject_constant) validate_config(config) return config def validate_config(config): if isinstance(config,dict) and 'infinbot_version' in config and 'foresixa_version' not in config: config=dict(config);config['foresixa_version']=config.pop('infinbot_version') if not isinstance(config, dict) or set(config) != {'foresixa_version', 'ingestion_lifecycle', 'active_worker_streams'}: raise Rejected('Invalid configuration structure') if not isinstance(config['foresixa_version'], str): raise Rejected('Invalid configuration version') ingress = config['ingestion_lifecycle'] if ingress != {'account_mode': 'PAPER_SANDBOX', 'verified_provider_endpoint': PAPER}: raise Rejected('Only the canonical paper account endpoint is supported') workers = config['active_worker_streams'] if not isinstance(workers, dict) or set(workers) != {'crypto_worker', 'futures_worker'}: raise Rejected('Explicit crypto and futures configuration required') crypto, futures = workers['crypto_worker'], workers['futures_worker'] if not isinstance(crypto, dict) or set(crypto) != {'stream_feed', 'symbols_monitored'}: raise Rejected('Invalid crypto configuration') symbols = crypto['symbols_monitored'] if (not isinstance(symbols, list) or not symbols or any(not isinstance(s, str) or s not in ('BTC/USD', 'ETH/USD', 'SOL/USD') for s in symbols) or len(set(symbols)) != len(symbols)): raise Rejected('Unsupported or duplicate crypto symbol') if not isinstance(crypto['stream_feed'], str): raise Rejected('Invalid declared feed label') if (not isinstance(futures, dict) or set(futures) != {'stream_feed', 'symbols_monitored', 'official_cme_point_multiplier'} or futures['symbols_monitored'] != ['/MES'] or not isinstance(futures['stream_feed'], str) or isinstance(futures['official_cme_point_multiplier'], bool) or number(futures['official_cme_point_multiplier'], True) != 5): raise Rejected('MES must declare a 5 USD per index point contract multiplier') def validate_quotes(payload, symbols, now): if not isinstance(payload, dict) or not isinstance(payload.get('quotes'), dict) or set(payload['quotes']) != set(symbols): raise Rejected('Missing requested quote observations') observations = {} for symbol in symbols: quote = payload['quotes'][symbol] if not isinstance(quote, dict): raise Rejected('Malformed quote') bid, ask = number(quote['bp'], True), number(quote['ap'], True) if bid > ask: raise Rejected('Crossed quote') at = instant(quote['t']) age = (now - at).total_seconds() if not -5 <= age <= 60: raise Rejected('Stale or future quote') observations[symbol] = {'bid': str(bid), 'ask': str(ask), 'provider_timestamp': at.isoformat(), 'age_seconds': age} return observations def read_json(url, headers): # Caller headers are deliberately not forwarded or logged. try:return read_proxy_json(url) except ProxyError as error:raise Rejected('Proxy read unavailable or rejected: '+str(error.http_status)) from None def reject_constant(value): raise Rejected('Nonfinite JSON rejected') def authenticate_account(get, headers): account = get(PAPER + '/v2/account', headers) if not isinstance(account, dict) or not isinstance(account.get('status'), str) or not account['status']: raise Rejected('Malformed account response') for field in ('cash', 'equity', 'buying_power'): account_amount(account.get(field)) if type(account.get('trading_blocked')) is not bool: raise Rejected('Missing account restriction flag') for field in ('account_blocked', 'transfers_blocked', 'trade_suspended_by_user'): if field in account and type(account[field]) is not bool: raise Rejected('Malformed restriction flag') # Report only the authentication result, never account identifiers/amounts. return 'PROXY_ACCOUNT_READ_VALIDATED' DEFAULT_CONFIG = {'foresixa_version': '1.0_READ_ONLY_FEED_DIAGNOSTIC', 'ingestion_lifecycle': {'account_mode': 'PAPER_SANDBOX', 'verified_provider_endpoint': 'https://sotality.com'}, 'active_worker_streams': {'crypto_worker': {'stream_feed': 'UNVERIFIED', 'symbols_monitored': ['BTC/USD', 'ETH/USD']}, 'futures_worker': {'stream_feed': 'UNCONFIGURED', 'symbols_monitored': ['/MES'], 'official_cme_point_multiplier': 5.0}}} class ForesixaLiveFeedVerifier: def __init__(self, config_path=None, *, config=None, get=read_json, clock=lambda: datetime.now(timezone.utc)): self.config_path = config_path self.config = config self.get, self.clock = get, clock def execute_exchange_stream_validation(self, *, settings=None, credentials_path=None, environment_only=False): report = { 'engine_architecture': 'INFINBOT_VALIDATION_CORE', 'execution_timestamp': self.clock().isoformat(), 'status': 'CONFIGURATION_BLOCKED', 'authentication_lifecycle': {'status': 'NOT_VERIFIED', 'account_mode': 'PROXY_UPSTREAM_MODE_UNVERIFIED'}, 'network_peripheral_gates': {'broker_handshake_clear': False, 'market_data_connectivity': 'NOT_VERIFIED', 'active_order_submission_allowed': False}, 'multi_worker_feed_registry': {'crypto_feed_status': 'NOT_VERIFIED', 'futures_feed_status': 'EXISTING_FUTURES_CONNECTION_NOT_VERIFIED_HERE', 'stocks_options_feed_status': 'NOT_CHECKED'}, 'stream_verified': False, 'broker_orders_submitted': 0, 'trial_started': False, 'live_clearance_granted': False, 'credential_rotation_verified': False, } try: config = self.config if self.config is not None else (load_worker_config(self.config_path) if self.config_path is not None else copy.deepcopy(DEFAULT_CONFIG)) validate_config(config) if settings is not None or credentials_path is not None or environment_only: raise Rejected('Client credential ingestion disabled; proxy owns upstream authentication') base, headers = account_credentials({}, 'paper') report['authentication_lifecycle']['account_mode']='PROXY_UPSTREAM_MODE_UNVERIFIED' report['proxy_base_url']=PROXY_BASE_URL report['broker_credential_files_read']=False report['direct_alpaca_calls']=0 report['credential_headers_sent']=False # Only the public worker config is read; no private keys or environment. # Public configuration cannot change the pinned proxy target. base, headers = account_credentials({}, 'paper') if base != config['ingestion_lifecycle']['verified_provider_endpoint']: raise Rejected('Endpoint mismatch') except Exception as error: report['reason'] = ('credentials.env is a folder; supply the actual private text file' if str(error) == 'CREDENTIAL_PATH_IS_FOLDER' else 'Private credentials.env file not found at the selected path' if str(error) == 'CREDENTIAL_FILE_MISSING' else 'Proxy-only configuration required; no client broker credential source is accepted') return json.dumps(report, indent=2, allow_nan=False) report['status'] = 'CHECKS_NOT_VERIFIED' try: auth_status = authenticate_account(self.get, headers) except Exception: auth_status = 'AUTHENTICATION_NOT_VERIFIED' authenticated = auth_status == 'PROXY_ACCOUNT_READ_VALIDATED' report['authentication_lifecycle']['status'] = auth_status report['network_peripheral_gates']['broker_handshake_clear'] = False report['network_peripheral_gates']['proxy_account_response_validated'] = authenticated symbols = config['active_worker_streams']['crypto_worker']['symbols_monitored'] quote_url = DATA + '/v1beta3/crypto/us/latest/quotes?' + urlencode({'symbols': ','.join(symbols)}) try: observations = validate_quotes(self.get(quote_url, headers), symbols, self.clock()) report['crypto_observations'] = {'source_endpoint': DATA, 'retrieved_at': self.clock().isoformat(), 'quotes': observations} report['network_peripheral_gates']['market_data_connectivity'] = 'FRESH_CRYPTO_REST_QUOTES_VERIFIED' report['multi_worker_feed_registry']['crypto_feed_status'] = 'FRESH_REST_QUOTES_VERIFIED' if authenticated: report['status'] = 'PARTIAL_PROXY_PAYLOAD_CHECKS_PASSED_EXISTING_FUTURES_UNVERIFIED' except Exception: report['crypto_quote_reason'] = 'Crypto request rejected, unavailable, incomplete, stale or malformed' # Never print credential fragments, broker account contents or arbitrary exception/server text. return json.dumps(report, indent=2, allow_nan=False) class ForesixaIntegratedVerifier(ForesixaLiveFeedVerifier): """Compatibility with the pasted two-path interface; paper and read-only.""" def __init__(self, config_path=None, credentials_path=None, **kwargs): super().__init__(config_path=config_path, **kwargs) self.credentials_path = credentials_path def execute_handshake_audit(self): return self.execute_exchange_stream_validation() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--config', help='Optional worker config JSON; omit for built-in BTC/USD and ETH/USD paper configuration') args = parser.parse_args() print(ForesixaIntegratedVerifier(args.config).execute_handshake_audit()) # Exit 1 means full multi-worker verification is incomplete; existing futures configuration is not inspected here. return 1 if __name__ == '__main__': raise SystemExit(main()) # Compatibility aliases preserve existing imports; the product is InFinBot. InFinBotLiveFeedVerifier=ForesixaLiveFeedVerifier InFinBotIntegratedVerifier=ForesixaIntegratedVerifier