from __future__ import annotations import json from typing import Any from sqlmodel import Session, select from ..connectors.portainer import PortainerConnector from ..connectors.unraid import UnraidConnector from ..core.config import settings from ..core.redaction import redact_message from ..core.time import utc_now from ..core.network_security import validate_connector_url from ..db.models import AppContainerMapping, AppSetting, AuditEvent, Connector, DiscoveryRun, Secret from ..schemas.forge_oidc import FORGE_OIDC_BINDING_SECRET_REF, ForgeOidcBindingV1 from .audit_logger import log_event from .secret_store import empty_secret_status, get_secret, has_secret, secret_status, secret_storage_status, set_secret from . import safety_center_service _CONNECTOR_FIELDS: dict[str, dict[str, Any]] = { 'portainer': { 'required': ['PORTAINER_URL', 'PORTAINER_API_KEY', 'PORTAINER_ENDPOINT_ID'], 'secret_env': 'PORTAINER_API_KEY', 'field_env': {'url': 'PORTAINER_URL', 'endpointId': 'PORTAINER_ENDPOINT_ID', 'verifyTls': 'PORTAINER_VERIFY_TLS'}, 'safe_fields': ['url', 'endpointId', 'verifyTls'], }, 'unraid': { 'required': ['UNRAID_URL', 'UNRAID_API_KEY'], 'secret_env': 'UNRAID_API_KEY', 'field_env': {'url': 'UNRAID_URL', 'verifyTls': 'UNRAID_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'sonarr': { 'required': ['SONARR_URL', 'SONARR_API_KEY'], 'secret_env': 'SONARR_API_KEY', 'field_env': {'url': 'SONARR_URL'}, 'safe_fields': ['url'], }, 'radarr': { 'required': ['RADARR_URL', 'RADARR_API_KEY'], 'secret_env': 'RADARR_API_KEY', 'field_env': {'url': 'RADARR_URL', 'verifyTls': 'RADARR_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'prowlarr': { 'required': ['PROWLARR_URL', 'PROWLARR_API_KEY'], 'secret_env': 'PROWLARR_API_KEY', 'field_env': {'url': 'PROWLARR_URL', 'verifyTls': 'PROWLARR_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'deluge': { 'required': ['DELUGE_URL', 'DELUGE_PASSWORD'], 'secret_env': 'DELUGE_PASSWORD', 'field_env': {'url': 'DELUGE_URL', 'verifyTls': 'DELUGE_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'plex': { 'required': ['PLEX_URL', 'PLEX_TOKEN'], 'secret_env': 'PLEX_TOKEN', 'field_env': {'url': 'PLEX_URL', 'verifyTls': 'PLEX_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'tautulli': { 'required': ['TAUTULLI_URL', 'TAUTULLI_API_KEY'], 'secret_env': 'TAUTULLI_API_KEY', 'field_env': {'url': 'TAUTULLI_URL', 'verifyTls': 'TAUTULLI_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'nginx_proxy_manager': { 'required': ['NPM_URL', 'NPM_IDENTITY', 'NPM_SECRET'], 'secret_env': 'NPM_SECRET', 'field_env': {'url': 'NPM_URL', 'identity': 'NPM_IDENTITY', 'verifyTls': 'NPM_VERIFY_TLS'}, 'safe_fields': ['url', 'identity', 'verifyTls'], }, 'authentik': { 'required': ['AUTHENTIK_URL', 'AUTHENTIK_TOKEN'], 'secret_env': 'AUTHENTIK_TOKEN', 'field_env': { 'url': 'AUTHENTIK_URL', 'verifyTls': 'AUTHENTIK_VERIFY_TLS', 'outpostUrl': 'AUTHENTIK_OUTPOST_URL', 'expectEmbeddedOutpost': 'AUTHENTIK_EXPECT_EMBEDDED_OUTPOST', }, 'safe_fields': ['url', 'verifyTls', 'outpostUrl', 'expectEmbeddedOutpost'], }, 'widefrog': { 'required': ['WIDEFROG_URL'], 'secret_env': 'WIDEFROG_API_KEY', 'field_env': {'url': 'WIDEFROG_URL', 'verifyTls': 'WIDEFROG_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, 'home_assistant': { 'required': ['HOME_ASSISTANT_URL', 'HOME_ASSISTANT_TOKEN'], 'secret_env': 'HOME_ASSISTANT_TOKEN', 'field_env': {'url': 'HOME_ASSISTANT_URL', 'verifyTls': 'HOME_ASSISTANT_VERIFY_TLS'}, 'safe_fields': ['url', 'verifyTls'], }, } def _setting_key(connector_id: str, suffix: str) -> str: return f'connector:{connector_id}:{suffix}' def _get_setting(session: Session, key: str) -> AppSetting | None: return session.get(AppSetting, key) def _set_setting(session: Session, key: str, value: dict[str, Any], is_secret: bool = False) -> AppSetting: row = _get_setting(session, key) if not row: row = AppSetting(key=key, is_secret=is_secret) row.value_json = value row.is_secret = is_secret row.updated_at = utc_now() session.add(row) return row def _connector_schema(connector_id: str) -> dict[str, Any]: return _CONNECTOR_FIELDS.get(connector_id, { 'required': [f'{connector_id.upper()}_URL'], 'secret_env': f'{connector_id.upper()}_API_KEY', 'field_env': {'url': f'{connector_id.upper()}_URL'}, 'safe_fields': ['url'], }) def _safe_config_from_row(session: Session, connector_id: str) -> dict[str, Any]: row = _get_setting(session, _setting_key(connector_id, 'config')) config = dict(row.value_json) if row else {} schema = _connector_schema(connector_id) safe: dict[str, Any] = {} for field in schema['safe_fields']: value = config.get(field) if value is None: env_name = schema['field_env'].get(field) value = settings.apis.get(env_name) if env_name else None if value is not None: safe[field] = value if connector_id == 'portainer' and not str(safe.get('endpointId') or '').strip(): safe['endpointId'] = '1' return safe def _has_secret(session: Session, connector_id: str) -> bool: schema = _connector_schema(connector_id) secret_row = _get_setting(session, _setting_key(connector_id, 'secret')) secret_env = schema.get('secret_env') return bool(secret_env and has_secret(session, secret_env)) or bool(secret_row and secret_row.value_json.get('value')) or bool(secret_env and settings.apis.get(secret_env)) def _connector_secret_status(session: Session, connector_id: str) -> dict[str, Any]: schema = _connector_schema(connector_id) secret_env = schema.get('secret_env') if not secret_env: return empty_secret_status(f'{connector_id.upper()}_API_KEY', used_by=[connector_id]) row = session.get(Secret, secret_env) if row: return secret_status(row) if settings.apis.get(secret_env): status = empty_secret_status(secret_env, used_by=[connector_id]) status['hasSecret'] = True status['status'] = 'env_fallback' status['lastUpdated'] = None return status return empty_secret_status(secret_env, used_by=[connector_id]) def _migrate_legacy_connector_secret(session: Session, connector_id: str) -> None: schema = _connector_schema(connector_id) secret_env = schema.get('secret_env') if not secret_env: return legacy_row = _get_setting(session, _setting_key(connector_id, 'secret')) legacy_value = legacy_row.value_json.get('value') if legacy_row else None if not isinstance(legacy_value, str) or not legacy_value.strip(): return set_secret( session, secret_env, legacy_value, used_by=[connector_id], metadata={'connectorId': connector_id, 'migratedFrom': _setting_key(connector_id, 'secret')}, commit=False, ) legacy_row.value_json = {'secretRef': secret_env, 'hasSecret': True, 'migrated': True} legacy_row.updated_at = utc_now() session.add(legacy_row) log_event( session, source='system', identity='secret_store', action_id='secret_migrated', target=secret_env, target_type='secret', risk='low', policy_decision='allowed', reason='Legacy plaintext connector secret migrated to encrypted secret store.', result='success', redacted_metadata={'connector': connector_id, 'secretRef': secret_env}, ) session.commit() def _apply_connector_config_to_runtime(session: Session, connector_id: str) -> None: _migrate_legacy_connector_secret(session, connector_id) schema = _connector_schema(connector_id) safe_config = _safe_config_from_row(session, connector_id) for field, env_name in schema['field_env'].items(): if field in safe_config: settings.apis[env_name] = safe_config[field] if 'verifyTls' in safe_config and connector_id in {'portainer', 'unraid'}: settings.verify_tls = bool(safe_config['verifyTls']) secret_row = _get_setting(session, _setting_key(connector_id, 'secret')) secret_env = schema.get('secret_env') if secret_env: stored_secret = get_secret(session, secret_env, touch=False) if stored_secret: settings.apis[secret_env] = stored_secret elif secret_row and secret_row.value_json.get('value'): settings.apis[secret_env] = secret_row.value_json.get('value') if connector_id == 'authentik': _apply_forge_oidc_binding_to_runtime(session) def _unique_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ValueError('duplicate_json_key') result[key] = value return result def _apply_forge_oidc_binding_to_runtime(session: Session) -> bool: """Load protected Forge resource IDs atomically and fail closed on bad state. This fixed secret reference is intentionally absent from connector setup fields and response schemas. Generic Authentik settings can only update the Authentik URL/TLS/token inputs, never these remote resource bindings. """ internal_keys = ( 'AUTHENTIK_FORGE_OIDC_PROVIDER_ID', 'AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING', 'AUTHENTIK_FORGE_OIDC_EMAIL_CLAIM_MAPPING', 'AUTHENTIK_FORGE_OIDC_PROFILE_CLAIM_MAPPING', ) settings.apis.update({key: None for key in internal_keys}) settings.apis['AUTHENTIK_FORGE_OIDC_BINDING_VALID'] = False try: raw = get_secret(session, FORGE_OIDC_BINDING_SECRET_REF, touch=False) if not isinstance(raw, str) or not raw: return False payload = json.loads(raw, object_pairs_hook=_unique_json_object) if ( not isinstance(payload, dict) or set(payload) != {'version', 'providerId', 'scopeMappingIds'} or not isinstance(payload.get('scopeMappingIds'), dict) or set(payload['scopeMappingIds']) != {'openid', 'email', 'profile'} ): return False binding = ForgeOidcBindingV1.model_validate(payload) except Exception: # noqa: BLE001 - binding failures never expose encrypted-store contents. return False # Only publish values after the whole versioned record and distinct UUIDs # have passed validation. Keep the stored JSON out of runtime settings. values = { 'AUTHENTIK_FORGE_OIDC_PROVIDER_ID': binding.provider_id, 'AUTHENTIK_FORGE_OIDC_OPENID_CLAIM_MAPPING': str(binding.scope_mapping_ids.openid), 'AUTHENTIK_FORGE_OIDC_EMAIL_CLAIM_MAPPING': str(binding.scope_mapping_ids.email), 'AUTHENTIK_FORGE_OIDC_PROFILE_CLAIM_MAPPING': str(binding.scope_mapping_ids.profile), } settings.apis.update(values) settings.apis['AUTHENTIK_FORGE_OIDC_BINDING_VALID'] = True return True def apply_stored_connector_settings(session: Session, connector_ids: set[str] | None = None) -> None: selected = connector_ids if connector_ids is not None else set(_CONNECTOR_FIELDS) for connector_id in _CONNECTOR_FIELDS: if connector_id not in selected: continue _apply_connector_config_to_runtime(session, connector_id) def serialize_connector_settings(session: Session, connector_id: str) -> dict[str, Any]: _apply_connector_config_to_runtime(session, connector_id) schema = _connector_schema(connector_id) missing = [name for name in schema['required'] if not settings.apis.get(name)] return { 'connectorId': connector_id, 'configured': not missing, 'hasSecret': _has_secret(session, connector_id), 'missing': missing, 'safeConfig': _safe_config_from_row(session, connector_id), 'secretRef': schema.get('secret_env'), 'secretStatus': _connector_secret_status(session, connector_id), 'secretStore': 'encrypted_local_db', 'secretStorage': secret_storage_status(), } def list_connector_settings(session: Session) -> list[dict[str, Any]]: connector_ids = [connector.id for connector in session.exec(select(Connector).order_by(Connector.id)).all()] for connector_id in _CONNECTOR_FIELDS: if connector_id not in connector_ids: connector_ids.append(connector_id) return [serialize_connector_settings(session, connector_id) for connector_id in connector_ids] def _normalize_connector_field(field: str, value: Any) -> Any: if field == 'verifyTls': return bool(value) if not isinstance(value, str): return value normalized = value.strip() if field in {'url', 'outpostUrl'}: normalized = validate_connector_url(normalized) return normalized def save_connector_settings(session: Session, connector_id: str, payload: dict[str, Any]) -> dict[str, Any]: schema = _connector_schema(connector_id) safe_payload = dict(_safe_config_from_row(session, connector_id)) for field in schema['safe_fields']: if field in payload and payload[field] is not None: safe_payload[field] = _normalize_connector_field(field, payload[field]) if connector_id == 'portainer' and not str(safe_payload.get('endpointId') or '').strip(): safe_payload['endpointId'] = '1' _set_setting(session, _setting_key(connector_id, 'config'), safe_payload, is_secret=False) secret_value = payload.get('apiKey') or payload.get('token') or payload.get('password') or payload.get('secret') if isinstance(secret_value, str) and secret_value.strip(): secret_env = schema.get('secret_env') if secret_env: set_secret( session, secret_env, secret_value, used_by=[connector_id], metadata={'connectorId': connector_id}, commit=False, ) _set_setting(session, _setting_key(connector_id, 'secret'), {'secretRef': secret_env, 'hasSecret': True}, is_secret=True) session.commit() _apply_connector_config_to_runtime(session, connector_id) connector = session.get(Connector, connector_id) if connector: connector.status = 'disconnected' if serialize_connector_settings(session, connector_id)['configured'] else 'missing_config' session.add(connector) session.commit() safety_center_service.invalidate_cache() return serialize_connector_settings(session, connector_id) def _step(step_id: str, label: str, status: str, required: bool) -> dict[str, Any]: return {'id': step_id, 'label': label, 'status': status, 'required': required} def _successful_portainer_test_exists(session: Session) -> bool: event = session.exec( select(AuditEvent) .where(AuditEvent.action_id == 'setup.test.portainer') .where(AuditEvent.result == 'success') .order_by(AuditEvent.timestamp.desc()) ).first() return bool(event) def _setup_flag(session: Session) -> tuple[bool, str | None]: first_run = _get_setting(session, 'setup:first_run_completed') completed_at = _get_setting(session, 'setup:completed_at') return bool(first_run and first_run.value_json.get('value')), completed_at.value_json.get('value') if completed_at else None def _set_setup_completed(session: Session, *, completed_at: str | None = None) -> str: now = completed_at or utc_now().isoformat() _set_setting(session, 'setup:first_run_completed', {'value': True}) _set_setting(session, 'setup:completed_at', {'value': now}) session.commit() safety_center_service.invalidate_cache() return now def _required_blockers( *, portainer_ready: bool, discovery_complete: bool, managed_apps: int, review_blockers: int, secret_storage_ready: bool, ) -> list[str]: blockers: list[str] = [] if not portainer_ready: blockers.append('portainer_not_connected') if not discovery_complete: blockers.append('discovery_not_completed') if managed_apps <= 0: blockers.append('no_managed_apps') if review_blockers > 0: blockers.append('apps_need_review') if not secret_storage_ready: blockers.append('secret_storage_not_ready') return blockers def get_setup_status(session: Session) -> dict[str, Any]: from .inventory_discovery import _latest_discovered_items from .app_registry import list_active_apps first_run_completed, setup_completed_at = _setup_flag(session) portainer = serialize_connector_settings(session, 'portainer') unraid = serialize_connector_settings(session, 'unraid') portainer_row = session.get(Connector, 'portainer') unraid_row = session.get(Connector, 'unraid') portainer_connected = portainer['configured'] and portainer_row is not None and portainer_row.status == 'connected' portainer_tested = _successful_portainer_test_exists(session) unraid_ready = unraid['configured'] and (unraid_row is None or unraid_row.status == 'connected') latest_run = session.exec(select(DiscoveryRun).order_by(DiscoveryRun.started_at.desc())).first() # Setup is an operator-facing current-state summary. Discovery run history # remains available under Inventory, but recreated Docker IDs must not make # the setup counters grow forever. discovered = _latest_discovered_items(session) mappings = session.exec(select(AppContainerMapping)).all() apps = list_active_apps(session) connectors = list_connector_settings(session) mapped_container_ids = {mapping.container_id for mapping in mappings} apps_need_review = len([app for app in apps if app.needs_review]) + len([ item for item in discovered if item.import_status in {'discovered', 'unmapped'} or (item.import_status == 'changed' and item.container_id not in mapped_container_ids) ]) optional_connectors = [item for item in connectors if item['connectorId'] != 'portainer'] missing_optional_sources = [item['connectorId'] for item in optional_connectors if item['missing']] missing_config = len([item for item in connectors if item['missing']]) mapped_apps = len(mappings) managed_apps = len(apps) review_blockers = 0 if first_run_completed else apps_need_review discovery_status = 'pending' if latest_run: discovery_status = 'complete' if latest_run.status == 'success' else 'failed' if latest_run.status == 'failed' else 'warning' discovery_complete = bool(latest_run and latest_run.status == 'success') or bool(discovered) or mapped_apps > 0 portainer_ready = portainer_connected or portainer_tested or discovery_complete review_status = 'pending' if not latest_run else ('complete' if review_blockers == 0 else 'warning') connectors_status = 'warning' if missing_optional_sources else 'complete' home_assistant_configured = 'home_assistant' not in missing_optional_sources voice_status = 'complete' if settings.voice_token and home_assistant_configured else 'warning' storage_status = secret_storage_status() secret_storage_ready = storage_status['status'] == 'ready' required_blockers = _required_blockers( portainer_ready=portainer_ready, discovery_complete=discovery_complete, managed_apps=mapped_apps, review_blockers=review_blockers, secret_storage_ready=secret_storage_ready, ) required_steps_complete = not required_blockers setup_complete = required_steps_complete and first_run_completed coverage_complete = not missing_optional_sources live_ready = setup_complete and settings.port == 1216 warnings: list[str] = [] if missing_optional_sources: warnings.append('Optional connector coverage is partial.') if settings.port != 1216: warnings.append('Runtime is not on canonical production port 1216.') next_actions = [ {'id': blocker, 'required': True} for blocker in required_blockers ] if required_steps_complete and not first_run_completed: next_actions.append({'id': 'reconcile_setup_state', 'required': True}) for connector_id in missing_optional_sources: next_actions.append({'id': f'configure_optional_{connector_id}', 'required': False}) steps = [ _step('portainer', 'Portainer verbinden', 'complete' if portainer_ready else 'pending', True), _step('unraid', 'Unraid API verbinden', 'complete' if unraid_ready else 'warning', False), _step('discovery', 'Container discovery draaien', 'complete' if discovery_complete else discovery_status, True), _step('review', 'Reviewbaseline vastgelegd' if first_run_completed else 'Apps reviewen', review_status, True), _step('connectors', 'Connectoren binden', connectors_status, False), _step('voice', 'Voice/Home Assistant instellen', voice_status, False), ] return { 'firstRunCompleted': first_run_completed, 'setupComplete': setup_complete, 'liveReady': live_ready, 'coverageComplete': coverage_complete, 'securityCoveragePartial': not coverage_complete, 'setupCompletedAt': setup_completed_at, 'requiredStepsComplete': required_steps_complete, 'optionalStepsComplete': coverage_complete, 'reviewBlockers': review_blockers, 'managedApps': managed_apps, 'needsReview': apps_need_review, 'missingOptionalSources': missing_optional_sources, 'steps': steps, 'warnings': warnings, 'nextActions': next_actions, 'blockers': required_blockers, 'summary': { 'containersDiscovered': len({item.container_id for item in discovered}), 'appsManaged': managed_apps, 'appsNeedReview': apps_need_review, 'connectorsMissingConfig': missing_config, 'missingOptionalSources': missing_optional_sources, }, } def complete_setup(session: Session) -> dict[str, Any]: from .inventory_discovery import sync_app_connector_statuses sync_app_connector_statuses(session) _set_setup_completed(session) return {'success': True, 'message': 'setup_completed', 'status': get_setup_status(session)} def reconcile_setup(session: Session) -> dict[str, Any]: from .inventory_discovery import sync_app_connector_statuses sync_app_connector_statuses(session) before = get_setup_status(session) blockers = list(before.get('blockers') or []) if blockers: log_event( session, source='manual', identity='setup', action_id='setup.reconcile', target='setup', target_type='setup', risk='low', policy_decision='allowed', reason='Setup reconciliation blocked by required criteria.', result='blocked', before_state={'firstRunCompleted': before['firstRunCompleted'], 'blockers': blockers}, after_state={'firstRunCompleted': before['firstRunCompleted']}, redacted_metadata={'blockers': blockers}, ) after = get_setup_status(session) return {'success': False, 'message': 'setup_reconcile_blocked', 'blockers': blockers, 'before': before, 'after': after} if not before['firstRunCompleted']: _set_setup_completed(session) after = get_setup_status(session) log_event( session, source='manual', identity='setup', action_id='setup.reconcile', target='setup', target_type='setup', risk='low', policy_decision='allowed', reason='Setup state reconciled from existing live data.', result='success', before_state={'firstRunCompleted': before['firstRunCompleted'], 'setupComplete': before['setupComplete']}, after_state={'firstRunCompleted': after['firstRunCompleted'], 'setupComplete': after['setupComplete']}, redacted_metadata={'managedApps': after['managedApps'], 'coverageComplete': after['coverageComplete']}, ) return {'success': True, 'message': 'setup_reconciled', 'blockers': [], 'before': before, 'after': after} def mark_setup_complete(session: Session, payload: dict[str, Any]) -> dict[str, Any]: if payload.get('confirm') != 'MARK_SETUP_COMPLETE': from fastapi import HTTPException raise HTTPException(status_code=400, detail='confirm_required') before = get_setup_status(session) _set_setup_completed(session) after = get_setup_status(session) log_event( session, source='manual', identity='setup', action_id='setup.mark_complete', target='setup', target_type='setup', risk='medium', policy_decision='allowed', reason='Operator explicitly marked setup complete.', result='success', before_state={'firstRunCompleted': before['firstRunCompleted'], 'setupComplete': before['setupComplete']}, after_state={'firstRunCompleted': after['firstRunCompleted'], 'setupComplete': after['setupComplete']}, redacted_metadata={'coverageComplete': after['coverageComplete'], 'blockers': after.get('blockers', [])}, ) return {'success': True, 'message': 'setup_marked_complete', 'status': after} def mark_voice_configured(session: Session) -> dict[str, Any]: _set_setting(session, 'setup:voice_configured', {'value': True}) session.commit() return {'success': True, 'message': 'voice_configured', 'status': get_setup_status(session)} def test_portainer_setup(session: Session) -> dict[str, Any]: config = serialize_connector_settings(session, 'portainer') if config['missing']: return {'success': False, 'status': 'missing_config', 'connected': False, 'requiredFields': config['missing'], 'message': 'missing_config'} connector = PortainerConnector() result = connector.test_connection() status = str(result.get('status', 'error')) if status != 'connected': message = redact_message(str(result.get('reason', status))) log_setup_test(session, 'portainer', status, message) response_status = 'unreachable' if status == 'error' else status required_fields = ['PORTAINER_ENDPOINT_ID'] if response_status == 'endpoint_error' else None return { 'success': False, 'status': response_status, 'connected': False, 'message': message, 'requiredFields': required_fields, 'safeConfig': config['safeConfig'], } containers = connector.list_containers() stacks = connector.list_stacks() if containers.get('status') != 'ok': message = redact_message(str(containers.get('reason', 'endpoint_error'))) log_setup_test(session, 'portainer', 'endpoint_error', message) return { 'success': False, 'status': 'endpoint_error', 'connected': False, 'message': message, 'requiredFields': ['PORTAINER_ENDPOINT_ID'], 'summary': {'containers': 0, 'stacks': 0, 'endpointReachable': False}, 'safeConfig': config['safeConfig'], } payload = { 'success': True, 'status': 'connected', 'connected': True, 'summary': { 'containers': len(containers.get('containers', [])), 'stacks': len(stacks.get('stacks', [])) if stacks.get('status') == 'ok' else 0, 'endpointReachable': True, }, 'safeConfig': config['safeConfig'], } connector_row = session.get(Connector, 'portainer') if connector_row: connector_row.status = 'connected' session.add(connector_row) session.commit() log_setup_test(session, 'portainer', 'connected', 'connected') return payload def test_unraid_setup(session: Session) -> dict[str, Any]: config = serialize_connector_settings(session, 'unraid') if config['missing']: return {'success': False, 'status': 'missing_config', 'connected': False, 'requiredFields': config['missing'], 'message': 'missing_config'} connector = UnraidConnector() result = connector.test_connection() status = str(result.get('status', 'error')) if status != 'connected': message = redact_message(str(result.get('reason', status))) log_setup_test(session, 'unraid', status, message) return {'success': False, 'status': 'unreachable' if status == 'error' else status, 'connected': False, 'message': message} info = connector.get_server_info() array = connector.get_array_status() docker = connector.get_docker_service_status() payload = { 'success': True, 'status': 'connected', 'connected': True, 'summary': { 'server': info.get('info', {}), 'arrayStatus': array.get('arrayStatus'), 'dockerService': docker.get('serviceStatus'), }, 'safeConfig': config['safeConfig'], } connector_row = session.get(Connector, 'unraid') if connector_row: connector_row.status = 'connected' session.add(connector_row) session.commit() log_setup_test(session, 'unraid', 'connected', 'connected') return payload def log_setup_test(session: Session, connector_id: str, result: str, reason: str) -> None: log_event( session, source='manual', identity='setup', action_id=f'setup.test.{connector_id}', target=connector_id, target_type='connector', risk='low', policy_decision='allowed', reason=redact_message(reason), result='success' if result == 'connected' else 'failed', redacted_metadata={'connector': connector_id, 'result': result}, )