"""Foresixa read-only account authentication check. Never prints credentials. This check cannot prove credential rotation, global secrecy, fresh quote connectivity, account review completion, or permission to trade. """ import argparse import json import os import shlex import stat from datetime import datetime, timezone from check_live_status import AlpacaLiveOnboardingAuditor, account_credentials, read_json, Rejected ALLOWED = frozenset({'ALPACA_BASE_URL', 'ALPACA_LIVE_KEY', 'ALPACA_LIVE_SECRET', 'ALPACA_PAPER_KEY', 'ALPACA_PAPER_SECRET'}) 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 class SecureCredentialValidator: def __init__(self, env_path='.env.production', *, mode='live', settings=None, environment_only=False, file_only=False, get=read_json): self.env_path = env_path self.mode = mode supplied = os.environ if settings is None else settings self.settings = {key: supplied[key] for key in ALLOWED if key in supplied} self.environment_only, self.file_only, self.get = environment_only, file_only, get self.last_report = None def authentication_report(self): report = { 'service': 'Foresixa Quant', 'observed_at': datetime.now(timezone.utc).isoformat(), 'mode': 'READ_ONLY_ACCOUNT_CHECK', 'account_mode': self.mode if self.mode in ('paper', 'live') else 'INVALID', 'status': 'CONFIGURATION_BLOCKED', 'credentials_in_output': False, 'credential_rotation_verified': False, 'fresh_market_feed_verified': False, 'futures_feed_verified': False, 'broker_orders_submitted': 0, 'trial_started': False, 'live_clearance_granted': False, } try: if self.mode not in ('paper', 'live') or self.environment_only and self.file_only: raise Rejected('Explicit matching account mode and one credential source required') prefix = 'ALPACA_LIVE_' if self.mode == 'live' else 'ALPACA_PAPER_' present = [bool(self.settings.get(prefix + suffix)) for suffix in ('KEY', 'SECRET')] # Never combine one environment credential with a possibly different file pair. if self.file_only: settings = load_private_credentials(self.env_path) elif any(present): if not all(present): raise Rejected('Complete credential pair required from one source') settings = self.settings elif self.environment_only: settings = self.settings else: settings = load_private_credentials(self.env_path) base, _ = account_credentials(settings, self.mode) except Exception: # Even an unexpected filesystem/parser error must not print a token or file contents. report['reason'] = 'Check account mode, complete fresh credentials, exact API endpoint and private file permissions' self.last_report = report return report report['endpoint'] = base report['status'] = 'AUTHENTICATION_NOT_VERIFIED' try: result = AlpacaLiveOnboardingAuditor(settings, self.mode, get=self.get).audit_onboarding_gate_status() if result.get('status') == 'ACCOUNT_READ_AUTHENTICATED': report['status'] = 'ACCOUNT_READ_AUTHENTICATED' else: report['reason'] = 'Read-only account request unavailable, rejected or malformed' except Exception: report['reason'] = 'Read-only account request unavailable, rejected or malformed' # No raw body, broker amounts, account identifiers, key fragments or exception text are returned. self.last_report = report return report def execute_blind_authentication_audit(self): report = self.authentication_report() print(json.dumps(report, indent=2, allow_nan=False)) return report['status'] == 'ACCOUNT_READ_AUTHENTICATED' def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--mode', choices=('paper', 'live'), required=True) source = parser.add_mutually_exclusive_group() source.add_argument('--env-file', help='Read only this private literal credential file') source.add_argument('--environment-only', action='store_true', help='Do not open a credential file') args = parser.parse_args() validator = SecureCredentialValidator(args.env_file or '.env.production', mode=args.mode, environment_only=args.environment_only, file_only=args.env_file is not None) return 0 if validator.execute_blind_authentication_audit() else 1 if __name__ == '__main__': raise SystemExit(main())