"""Read a local licensing registry; never provision tenants, bill or enable trading.""" from data_paths import default_db_path import argparse import json import math import sqlite3 import tempfile from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from pathlib import Path ROOT = Path(__file__).resolve().parent DEFAULT_DB = str(default_db_path(ROOT / 'foresixa_commercial_fleet.db')) REQUIRED = {'tenant_id', 'client_name', 'license_key', 'license_status', 'client_custom_risk_pct', 'client_custom_drawdown_limit', 'monthly_lease_fee', 'activated_at', 'expires_at'} def amount(value): if isinstance(value, bool): raise ValueError('Finite nonnegative number required') try: result = Decimal(str(value)) except (InvalidOperation, ValueError): raise ValueError('Finite nonnegative number required') from None if not result.is_finite() or result < 0: raise ValueError('Finite nonnegative number required') return result def timestamp(value): try: result = datetime.fromisoformat(value.replace('Z', '+00:00')) except (ValueError, AttributeError, TypeError): raise ValueError('Timezone-aware ISO timestamp required') from None if result.tzinfo is None or result.utcoffset() is None: raise ValueError('Timezone-aware ISO timestamp required') return result.astimezone(timezone.utc) def initialize_registry(db_path): """Explicit schema-only initialization in a fresh, private, dedicated file.""" path = Path(db_path).resolve() with path.open('xb'): pass path.chmod(0o600) conn = sqlite3.connect(path) try: conn.execute('PRAGMA journal_mode=WAL') conn.execute('PRAGMA synchronous=FULL') conn.executescript((ROOT / 'init_commercial_tenants.sql').read_text()) conn.commit() finally: conn.close() class FleetDashboard: """Local administrator report, not an authenticated multi-tenant API.""" def __init__(self, db_path=DEFAULT_DB, *, now=None): path = Path(db_path).resolve() if not path.is_file(): raise FileNotFoundError('Existing commercial registry required; no file created') self.conn = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=1) self.conn.row_factory = sqlite3.Row self.conn.execute('PRAGMA query_only=ON') self.now = timestamp((now or datetime.now(timezone.utc)).isoformat()) def _parse_tenant(self, row): record = {'tenant_id': row['tenant_id'], 'client_name': row['client_name'], 'recorded_license_status': row['license_status'], 'license_key': '[REDACTED]', 'effective_license_status': 'INVALID_RECORD', 'eligibility_diagnostic': 'BLOCKED', 'trading_enabled': False, 'worker_parameters_applied': False} try: if type(row['tenant_id']) is not int or row['tenant_id'] <= 0: raise ValueError('Positive tenant ID required') if not isinstance(row['client_name'], str) or not row['client_name'].strip(): raise ValueError('Client name required') if not row['license_present']: raise ValueError('Nonempty license key required') if row['license_status'] not in ('ACTIVE', 'SUSPENDED', 'TRIAL'): raise ValueError('Unknown license status') risk = amount(row['client_custom_risk_pct']) drawdown = amount(row['client_custom_drawdown_limit']) fee = amount(row['monthly_lease_fee']) activated = timestamp(row['activated_at']) expires = timestamp(row['expires_at']) if expires <= activated: raise ValueError('Expiry must follow activation') effective = row['license_status'] if self.now >= expires: effective = 'EXPIRED' elif self.now < activated: effective = 'NOT_YET_ACTIVE' config_valid = risk <= 1 and drawdown <= 3 diagnostic = 'BLOCKED' if config_valid and effective == 'ACTIVE': diagnostic = 'LOCAL_LICENSE_RECORD_ELIGIBLE_ONLY' elif config_valid and effective == 'TRIAL': diagnostic = 'TRIAL_RESEARCH_ONLY' record.update(effective_license_status=effective, eligibility_diagnostic=diagnostic, risk_configuration_status='WITHIN_RESEARCH_CAPS' if config_valid else 'RISK_CONFIGURATION_BLOCKED', client_custom_risk_pct=format(risk, 'f'), client_custom_drawdown_limit_pct=format(drawdown, 'f'), monthly_lease_fee_as_recorded=format(fee, 'f'), activated_at=activated.isoformat(), expires_at=expires.isoformat()) except ValueError as exc: record['validation_error'] = str(exc) return record def generate_summary(self, tenant_id=None): if tenant_id is not None and (type(tenant_id) is not int or tenant_id <= 0): raise ValueError('Positive explicit tenant ID required') report = {'status': 'REGISTRY_UNAVAILABLE', 'reported_at': self.now.isoformat(), 'scope': {'tenant_id': tenant_id} if tenant_id else {'local_administrator': 'ALL_TENANTS'}, 'tenants': [], 'total_tenants_in_scope': 0, 'billing_executed': False, 'worker_parameters_applied': False, 'broker_orders_submitted': 0, 'live_clearance_granted': False} try: self.conn.execute('BEGIN') if [r[0] for r in self.conn.execute('PRAGMA quick_check')] != ['ok']: raise ValueError('Database structural check failed') columns = {r[1] for r in self.conn.execute('PRAGMA table_info(commercial_tenant_licenses)')} if not REQUIRED <= columns: raise ValueError('Missing or incompatible commercial_tenant_licenses table') # Key bytes are deliberately never selected into a report. query = '''SELECT tenant_id, client_name, (typeof(license_key) = 'text' AND length(trim(license_key)) > 0) AS license_present, license_status, client_custom_risk_pct, client_custom_drawdown_limit, monthly_lease_fee, activated_at, expires_at FROM commercial_tenant_licenses''' values = () if tenant_id is not None: query += ' WHERE tenant_id = ?' values = (tenant_id,) rows = self.conn.execute(query + ' ORDER BY tenant_id', values).fetchall() report['tenants'] = [self._parse_tenant(row) for row in rows] counts = {} for row in report['tenants']: status = row['effective_license_status'] counts[status] = counts.get(status, 0) + 1 complete_status = 'FLEET_REPORT_WITH_INVALID_RECORDS' if counts.get('INVALID_RECORD') else 'FLEET_REPORT_COMPLETE' report.update(status=complete_status if rows else ('TENANT_NOT_FOUND' if tenant_id else 'REGISTRY_EMPTY'), total_tenants_in_scope=len(rows), effective_status_counts=counts) return report except (ValueError, sqlite3.Error): report['reason'] = 'Missing, incompatible or unreadable commercial registry; no eligibility granted' return report finally: self.conn.rollback() def close(self): self.conn.close() class SovereignFleetDashboardController(FleetDashboard): """Supplied controller interface; registration is an in-memory fixture only.""" def __init__(self, db_filepath=DEFAULT_DB, *, now=None): self.fixture_mode = str(db_filepath) == ':memory:' self.last_registration_result = None if not self.fixture_mode: super().__init__(db_filepath, now=now) return self.conn = sqlite3.connect(':memory:') self.conn.row_factory = sqlite3.Row self.conn.executescript((ROOT / 'init_commercial_tenants.sql').read_text()) self.now = timestamp((now or datetime.now(timezone.utc)).isoformat()) def register_new_commercial_lease(self, client, license_str, max_risk, max_dd, lease_fee): if not self.fixture_mode: self.last_registration_result = 'BLOCKED_READ_ONLY_REGISTRY' return False try: if not all(isinstance(value, str) and value.strip() for value in (client, license_str)): raise ValueError('Client and license key required') risk, drawdown, fee = amount(max_risk), amount(max_dd), amount(lease_fee) if risk > 1 or drawdown > 3: raise ValueError('Research risk caps exceeded') if not math.isfinite(float(fee)): raise ValueError('Fee cannot be represented in the supplied REAL schema') # Decimal inputs are validated first; REAL storage follows the supplied schema. with self.conn: cursor = self.conn.execute('''INSERT INTO commercial_tenant_licenses (client_name,license_key,license_status,client_custom_risk_pct, client_custom_drawdown_limit,monthly_lease_fee,activated_at,expires_at) VALUES (?,?,'ACTIVE',?,?,?,?,?) ON CONFLICT(license_key) DO NOTHING''', (client,license_str,str(risk),str(drawdown),str(fee),self.now.isoformat(), (self.now + timedelta(days=30)).isoformat())) inserted = cursor.rowcount == 1 self.last_registration_result = 'SIMULATED_RECORD_INSERTED' if inserted else 'DUPLICATE_KEY_NOT_INSERTED' return inserted except (ValueError, sqlite3.Error): self.conn.rollback() self.last_registration_result = 'INVALID_REGISTRATION_INPUT' return False def compile_desktop_mobile_dashboard_feed(self, tenant_id=None): result = self.generate_summary(tenant_id) active = [row for row in result['tenants'] if row['eligibility_diagnostic'] == 'LOCAL_LICENSE_RECORD_ELIGIBLE_ONLY'] result.update(engine_status='SIMULATED_FLEET_ONLY' if self.fixture_mode else 'LOCAL_ADMINISTRATOR_REPORT_ONLY', global_analytics={'active_license_records_in_scope':len(active), 'collected_revenue_verified':False}) if self.fixture_mode: result.update(input_provenance='SYNTHETIC_REGISTRATION_FIXTURE_ONLY', production_evidence=False) result['global_analytics']['simulated_monthly_fee_schedule_amount'] = format( sum((amount(row['monthly_lease_fee_as_recorded']) for row in active),Decimal('0')),'f') result['global_analytics']['currency_assumption'] = 'USD_AS_SUPPLIED_FOR_FIXTURE_ONLY' return result def lease_fixture_report(): controller = SovereignFleetDashboardController(':memory:') try: inserted = [ controller.register_new_commercial_lease('Client_Alpha_Individual', 'KEY_X9921_ALPHA', .5, 2, 1500), controller.register_new_commercial_lease('Sovereign_Capital_Fund', 'KEY_Z8812_INSTITUTION', 1, 3, 12500)] result = controller.compile_desktop_mobile_dashboard_feed() result['supplied_fixture_registrations_inserted'] = sum(inserted) return result finally: controller.close() def fixture_report(): """Temporary fabricated tenants, separate from every persistent registry.""" with tempfile.TemporaryDirectory(prefix='foresixa-fleet-fixture-') as directory: db = Path(directory) / 'fixture.db' initialize_registry(db) with sqlite3.connect(db) as conn: conn.executemany('''INSERT INTO commercial_tenant_licenses (client_name, license_key, license_status, monthly_lease_fee, activated_at, expires_at) VALUES (?, ?, ?, ?, ?, ?)''', [ ('Fixture Active', 'FIXTURE_ONLY_ACTIVE', 'ACTIVE', 100, '2026-10-01T00:00:00Z', '2026-11-01T00:00:00Z'), ('Fixture Trial', 'FIXTURE_ONLY_TRIAL', 'TRIAL', 0, '2026-10-01T00:00:00Z', '2026-11-01T00:00:00Z'), ('Fixture Suspended', 'FIXTURE_ONLY_SUSPENDED', 'SUSPENDED', 100, '2026-10-01T00:00:00Z', '2026-11-01T00:00:00Z'), ('Fixture Expired', 'FIXTURE_ONLY_EXPIRED', 'ACTIVE', 100, '2026-09-01T00:00:00Z', '2026-10-01T00:00:00Z'), ('Fixture Future', 'FIXTURE_ONLY_FUTURE', 'ACTIVE', 100, '2026-11-01T00:00:00Z', '2026-12-01T00:00:00Z')]) dashboard = FleetDashboard(db, now=datetime(2026, 10, 5, tzinfo=timezone.utc)) try: result = dashboard.generate_summary() result['input_provenance'] = 'SYNTHETIC_PARSING_FIXTURE_ONLY' result['production_evidence'] = False return result finally: dashboard.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--db-path', default=str(DEFAULT_DB)) parser.add_argument('--tenant-id', type=int) parser.add_argument('--initialize-only', action='store_true') parser.add_argument('--demo', action='store_true') parser.add_argument('--verify-registration', action='store_true') args = parser.parse_args() if args.demo and (args.initialize_only or args.tenant_id is not None or args.db_path != str(DEFAULT_DB)): parser.error('--demo cannot initialize or inspect a tenant registry') if args.initialize_only and args.tenant_id is not None: parser.error('--initialize-only cannot inspect a tenant') if args.verify_registration and (args.demo or args.initialize_only or args.tenant_id is not None or args.db_path != str(DEFAULT_DB)): parser.error('--verify-registration uses only an isolated in-memory fixture') dashboard = None try: if args.verify_registration: result = lease_fixture_report() elif args.demo: result = fixture_report() elif args.initialize_only: initialize_registry(args.db_path) result = {'status': 'EMPTY_COMMERCIAL_SCHEMA_CREATED', 'tenants_seeded': 0, 'billing_executed': False, 'live_clearance_granted': False} else: dashboard = FleetDashboard(args.db_path) result = dashboard.generate_summary(args.tenant_id) print(json.dumps(result, indent=2, allow_nan=False)) return 1 if result['status'] in ('REGISTRY_UNAVAILABLE', 'TENANT_NOT_FOUND', 'FLEET_REPORT_WITH_INVALID_RECORDS') else 0 except (OSError, ValueError, sqlite3.Error): print(json.dumps({'status': 'REGISTRY_UNAVAILABLE', 'reason': 'Existing readable registry required, or choose a fresh path for explicit schema initialization', 'live_clearance_granted': False})) return 1 finally: if dashboard: dashboard.close() if __name__ == '__main__': raise SystemExit(main())