ServerKit/backend/app/services/drift_service.py

"""Configuration drift detection + explicit per-resource repair.

ServerKit's service layer is imperative: it writes nginx vhosts and compose
override files from DB state and assumes they stay put. This module closes the
loop *gradually*: it re-renders what each managed file **should** contain
(through the exact same renderers the panel uses when writing them), compares
against what is actually on disk, and reports the difference. Nothing here
auto-repairs — :meth:`DriftService.repair` runs only when explicitly invoked
(API / doctor batch), per single resource.

Check framework
---------------
``DRIFT_CHECKS`` maps a resource ``type`` to a small check dict:

    {
        'type': 'nginx_vhost',
        'title': 'Nginx vhosts',
        'supported': fn() -> (bool, reason),          # optional platform gate
        'list_resources': fn() -> [(id, name)],
        'render_expected': fn(resource_id) -> {path: content|None},
        'read_actual': fn(paths) -> {path: content|None},  # optional; default
                                                           # reads the files
        'repair': fn(resource_id) -> {'success', 'wrote', 'reloaded', ...},
    }

``render_expected`` values: a string is the exact content the file must have;
``None`` means the file must NOT exist (e.g. a stale compose override that
should have been removed). An empty dict means nothing applies to this
resource → it reports ``in_sync``.

Platform policy (documented choice): the built-in checks cover Linux server
paths (``/etc/nginx/...``, app roots). On a non-Linux host each built-in check
reports a single ``status='error'`` entry with detail ``'unsupported on this
host'`` — we deliberately do NOT report ``in_sync`` there, because "we couldn't
look" is not "it matches". The gate lives per-check (``supported``) so tests
and future cross-platform checks run anywhere.

Systemd units: the panel *does* template gunicorn units
(``PythonService.create_gunicorn_service``), but their render inputs (worker
count, run-as user, and a point-in-time snapshot of the app's ``.env``) are
not persisted anywhere, so a faithful expected-render is impossible —
re-rendering with today's defaults would flag false drift. The ServerKit
panel/agent units themselves are installer-owned. No systemd drift check is
registered for now.
"""
import difflib
import json
import logging
import os
import re
import sys
from datetime import datetime

logger = logging.getLogger(__name__)

DRIFT_JOB_KIND = 'drift.check'
DRIFT_SCHEDULE_NAME = 'drift-check'
LAST_REPORT_KEY = 'drift_last_report'
DIFF_MAX_LINES = 200

# type -> check dict (see module docstring for the shape).
DRIFT_CHECKS = {}


def register_check(check):
    """Register (or replace) a drift check. Plugins may add their own."""
    DRIFT_CHECKS[check['type']] = check
    return check


def _is_linux():
    return sys.platform.startswith('linux')


def _linux_only():
    if _is_linux():
        return True, None
    return False, 'unsupported on this host'


def _default_read_actual(paths):
    """Read each path from disk; ``None`` for a file that doesn't exist."""
    actual = {}
    for path in paths:
        try:
            if os.path.exists(path):
                with open(path, 'r', encoding='utf-8', errors='replace') as f:
                    actual[path] = f.read()
            else:
                actual[path] = None
        except OSError as e:
            raise RuntimeError(f'cannot read {path}: {e}')
    return actual


def _unified_diff(path, actual, expected):
    """Unified diff actual → expected for one file, as a list of lines."""
    a = (actual or '').splitlines()
    b = (expected or '').splitlines()
    return list(difflib.unified_diff(
        a, b,
        fromfile=f'{path} (actual)',
        tofile=f'{path} (expected)',
        lineterm='',
    ))


class DriftService:
    """Detect + repair drift between DB-derived config and files on disk."""

    # ------------------------------------------------------------------ #
    # Detection
    # ------------------------------------------------------------------ #

    @classmethod
    def check_resource(cls, check, resource_id, name):
        """Run one check against one resource. Returns a report entry."""
        entry = {
            'type': check['type'],
            'id': resource_id,
            'name': name,
            'status': 'in_sync',
            'diff': None,
            'checked_at': datetime.utcnow().isoformat() + 'Z',
        }
        try:
            expected = check['render_expected'](resource_id) or {}
            if not expected:
                # Nothing applies to this resource (e.g. app publishes no
                # files) — by definition nothing can have drifted.
                return entry
            reader = check.get('read_actual') or _default_read_actual
            actual = reader(list(expected.keys()))

            diff_lines = []
            statuses = set()
            for path, want in expected.items():
                have = actual.get(path)
                if want is None:
                    # File must not exist.
                    if have is None:
                        statuses.add('in_sync')
                    else:
                        statuses.add('drifted')
                        diff_lines += _unified_diff(path, have, '')
                elif have is None:
                    statuses.add('missing')
                    diff_lines += _unified_diff(path, '', want)
                elif have == want:
                    statuses.add('in_sync')
                else:
                    statuses.add('drifted')
                    diff_lines += _unified_diff(path, have, want)

            if 'drifted' in statuses:
                entry['status'] = 'drifted'
            elif 'missing' in statuses:
                entry['status'] = 'missing'
            if diff_lines:
                if len(diff_lines) > DIFF_MAX_LINES:
                    kept = diff_lines[:DIFF_MAX_LINES]
                    kept.append(
                        f'... (diff truncated at {DIFF_MAX_LINES} lines)')
                    diff_lines = kept
                entry['diff'] = '\n'.join(diff_lines)
        except Exception as e:  # noqa: BLE001 — one bad resource must not kill the sweep
            entry['status'] = 'error'
            entry['detail'] = str(e)
        return entry

    @classmethod
    def check_all(cls):
        """Run every registered check against every resource.

        Returns a list of entries ``{type, id, name, status, diff,
        checked_at}`` with ``status`` in
        ``in_sync | drifted | missing | error``.
        """
        results = []
        for check in DRIFT_CHECKS.values():
            supported = check.get('supported')
            if supported:
                ok, reason = supported()
                if not ok:
                    results.append({
                        'type': check['type'],
                        'id': None,
                        'name': check.get('title', check['type']),
                        'status': 'error',
                        'diff': None,
                        'detail': reason or 'unsupported on this host',
                        'checked_at': datetime.utcnow().isoformat() + 'Z',
                    })
                    continue
            try:
                resources = check['list_resources']()
            except Exception as e:  # noqa: BLE001
                results.append({
                    'type': check['type'],
                    'id': None,
                    'name': check.get('title', check['type']),
                    'status': 'error',
                    'diff': None,
                    'detail': str(e),
                    'checked_at': datetime.utcnow().isoformat() + 'Z',
                })
                continue
            for resource in resources:
                if isinstance(resource, dict):
                    # list_resources may report a per-resource probe failure
                    # inline (an app it could not even inspect) — pass it
                    # through as an error entry instead of dropping the app.
                    resource.setdefault('type', check['type'])
                    resource.setdefault('checked_at',
                                        datetime.utcnow().isoformat() + 'Z')
                    results.append(resource)
                    continue
                resource_id, name = resource
                results.append(cls.check_resource(check, resource_id, name))
        return results

    # ------------------------------------------------------------------ #
    # Last-report storage
    # ------------------------------------------------------------------ #

    @classmethod
    def build_report(cls):
        results = cls.check_all()
        return {
            'results': results,
            'drifted': sum(1 for r in results if r['status'] in ('drifted', 'missing')),
            'errors': sum(1 for r in results if r['status'] == 'error'),
            'generated_at': datetime.utcnow().isoformat() + 'Z',
        }

    @classmethod
    def store_report(cls, report):
        from app.services.settings_service import SettingsService
        SettingsService.set(LAST_REPORT_KEY, json.dumps(report))

    @classmethod
    def get_last_report(cls):
        from app.services.settings_service import SettingsService
        raw = SettingsService.get(LAST_REPORT_KEY)
        if not raw:
            return None
        try:
            return json.loads(raw)
        except (TypeError, ValueError):
            return None

    # ------------------------------------------------------------------ #
    # Repair (explicit, per-resource — never automatic)
    # ------------------------------------------------------------------ #

    @classmethod
    def repair(cls, check_type, resource_id):
        """Re-render + rewrite one resource through the panel's normal write
        path (incl. nginx reload / compose up where applicable).

        Returns ``{'success': bool, 'wrote': [paths], 'reloaded': bool}`` (plus
        ``'error'`` on failure). Strictly on-demand.
        """
        check = DRIFT_CHECKS.get(check_type)
        if not check:
            return {'success': False, 'error': f'Unknown drift check type: {check_type}',
                    'wrote': [], 'reloaded': False}
        supported = check.get('supported')
        if supported:
            ok, reason = supported()
            if not ok:
                return {'success': False, 'error': reason or 'unsupported on this host',
                        'wrote': [], 'reloaded': False}
        repair_fn = check.get('repair')
        if not repair_fn:
            return {'success': False, 'error': f'{check_type} is not repairable',
                    'wrote': [], 'reloaded': False}
        try:
            result = repair_fn(resource_id) or {}
        except Exception as e:  # noqa: BLE001 — surface as a clean error
            return {'success': False, 'error': str(e), 'wrote': [], 'reloaded': False}
        result.setdefault('success', False)
        result.setdefault('wrote', [])
        result.setdefault('reloaded', False)
        return result

    # ------------------------------------------------------------------ #
    # Job plumbing
    # ------------------------------------------------------------------ #

    @classmethod
    def run_drift_check_job(cls, job):
        """Job handler for ``drift.check``: sweep, store, notify on drift."""
        report = cls.build_report()
        cls.store_report(report)
        if report['drifted'] > 0:
            cls._notify_drift(report)
        return {'drifted': report['drifted'], 'errors': report['errors'],
                'checked': len(report['results'])}

    @classmethod
    def _notify_drift(cls, report):
        """Best-effort admin notification — never breaks the sweep."""
        try:
            from app.plugins_sdk import notify
            names = [r['name'] for r in report['results']
                     if r['status'] in ('drifted', 'missing')]
            summary = ', '.join(names[:5]) + ('…' if len(names) > 5 else '')
            notify.send('drift.detected', to='admins', data={
                'count': report['drifted'],
                'resources': names,
                'message': (
                    f"{report['drifted']} managed configuration file(s) no longer match "
                    f"what ServerKit expects: {summary}. Review them on "
                    'Monitoring → Doctor and repair the ones you did not change '
                    'on purpose.'),
            })
        except Exception as e:  # noqa: BLE001
            logger.debug('drift notification failed: %s', e)

    @classmethod
    def register_jobs(cls):
        """Register the drift sweep handler with the job registry.
        Called once at app startup (see app/__init__.py)."""
        from app.jobs import registry
        registry.register(DRIFT_JOB_KIND, cls.run_drift_check_job, replace=True)


# --------------------------------------------------------------------------- #
# Built-in checks
# --------------------------------------------------------------------------- #

def _nginx_list_resources():
    """Apps published via host nginx = apps that have Domain rows."""
    from app.models.application import Application
    from app.models.domain import Domain
    # Both sides must be live: drift REPAIR rewrites the vhost, so a tombstoned
    # app (or an app whose only remaining domains are tombstoned) would have the
    # config the delete removed written back.
    rows = (Application.query_active()
            .join(Domain, Domain.application_id == Application.id)
            .filter(Domain.deleted_at.is_(None))
            .distinct()
            .all())
    return [(app.id, app.name) for app in rows]


def _nginx_render_expected(app_id):
    from app.models.application import Application
    from app.services.nginx_service import NginxService
    from app.services.site_domain_service import SiteDomainService

    app = Application.query_active().filter_by(id=app_id).first()
    if app is None:
        return {}
    kwargs, warning = SiteDomainService.app_vhost_kwargs(app)
    if kwargs is None:
        if warning:
            # App has domains but can't be published via host nginx — there is
            # nothing the panel would have written, so nothing to compare.
            logger.debug('drift: nginx vhost n/a for %s: %s', app.name, warning)
        return {}
    rendered = NginxService.render_site_config(**kwargs)
    if not rendered.get('success'):
        raise RuntimeError(rendered.get('error') or 'vhost render failed')
    config = rendered['config']

    # Feature blocks live in the vhost but not in the template: an app with
    # an enforcing WAF policy has its include wired post-render
    # (WafService.apply). Render the expected file through the SAME injector,
    # or every WAF'd app reads as permanently drifted and one "repair" click
    # regenerates the vhost without the include — silently unprotecting the
    # site while the DB still says enforcing.
    config = _with_expected_waf(app.id, config)

    path = os.path.join(NginxService.SITES_AVAILABLE, app.name)
    return {path: config}


def _with_expected_waf(app_id, config):
    from app.models.waf_policy import WafPolicy
    from app.services.waf_service import WafService

    policy = WafPolicy.query.filter_by(application_id=app_id).first()
    if policy is None or policy.mode not in ('block', 'detect'):
        return config
    injected = WafService.inject_include_text(
        config, WafService.include_path_for(app_id))
    return injected if injected is not None else config


# A vhost written before the timed log format has a bare
# `access_log <path>;`. That is not drift: the format only changes what the log
# records, and the next regeneration adopts it. Read such lines as the timed
# form so an upgrade doesn't flag (and notify about) every existing site.
_LEGACY_ACCESS_LOG_RE = re.compile(
    r'^(\s*access_log /var/log/nginx/[^\s;]+\.access\.log);$', re.MULTILINE)
# Older vhosts carry an internal reference after this comment; the comment is
# all that changed, so it is not drift either.
_LEGACY_ASSET_COMMENT_RE = re.compile(
    r'^(\s*# ServerKit: fingerprinted assets never change) \([^)\n]*\)$', re.MULTILINE)


def _nginx_read_actual(paths):
    actual = _default_read_actual(paths)
    return {
        path: (_LEGACY_ASSET_COMMENT_RE.sub(
                   r'\1', _LEGACY_ACCESS_LOG_RE.sub(r'\1 serverkit_timed;', content))
               if content is not None else None)
        for path, content in actual.items()
    }


def _nginx_repair(app_id):
    from app.models.application import Application
    from app.services.nginx_service import NginxService
    from app.services.site_domain_service import SiteDomainService

    app = Application.query_active().filter_by(id=app_id).first()
    if app is None:
        return {'success': False, 'error': 'Service not found'}
    res = SiteDomainService.write_app_vhost(app)
    if res.get('warning') and not (res.get('nginx') or {}).get('success'):
        return {'success': False, 'error': res['warning']}

    # write_app_vhost regenerates from the template, which drops the WAF
    # include; re-wire it for apps with an enforcing policy so a repair
    # never strips protection the DB still records.
    from app.services.waf_service import WafService
    WafService.reapply_if_enforcing(app_id)

    reload_res = NginxService.reload()
    return {
        'success': True,
        'wrote': [os.path.join(NginxService.SITES_AVAILABLE, app.name)],
        'reloaded': bool(reload_res.get('success')),
        'warning': res.get('warning') or (None if reload_res.get('success')
                                          else reload_res.get('error')),
    }


register_check({
    'type': 'nginx_vhost',
    'title': 'Nginx vhosts',
    'supported': _linux_only,
    'list_resources': _nginx_list_resources,
    'render_expected': _nginx_render_expected,
    'read_actual': _nginx_read_actual,
    'repair': _nginx_repair,
})


def _list_error_entry(check_type, resource_id, name, detail):
    """An inline 'could not check this resource' entry for list_resources to
    return alongside healthy (id, name) tuples — same shape check_all builds
    for a whole-check failure, so a resource one app could not inspect is
    reported as an error instead of vanishing from the sweep."""
    return {
        'type': check_type,
        'id': resource_id,
        'name': name,
        'status': 'error',
        'diff': None,
        'detail': detail,
        'checked_at': datetime.utcnow().isoformat() + 'Z',
    }


def _compose_list_resources():
    """Managed apps that live in a project dir (candidate compose apps)."""
    from app.models.application import Application
    rows = Application.query_active().filter(Application.root_path.isnot(None)).all()
    out = []
    for app in rows:
        try:
            from app.services.compose_env_service import ComposeEnvService
            if app.root_path and os.path.isdir(app.root_path) and \
                    ComposeEnvService.find_base_compose(app.root_path, app.compose_file):
                out.append((app.id, app.name))
        except Exception as e:  # noqa: BLE001 — an unreadable app is an error, not an absence
            out.append(_list_error_entry(
                'compose_override', app.id, app.name,
                f'Could not inspect this app: {e}'))
    return out


def _compose_render_expected(app_id):
    from app.models.application import Application
    from app.services.compose_env_service import ComposeEnvService

    app = Application.query_active().filter_by(id=app_id).first()
    if app is None or not app.root_path:
        return {}
    spec = ComposeEnvService.render_override(app.root_path, app.compose_file)
    if not spec['applies']:
        return {}
    # content None → the override must not exist (stale file = drift).
    return {spec['path']: spec['content']}


def _compose_repair(app_id):
    from app.models.application import Application
    from app.services.compose_env_service import ComposeEnvService
    from app.services.docker_service import DockerService

    app = Application.query_active().filter_by(id=app_id).first()
    if app is None:
        return {'success': False, 'error': 'Service not found'}
    written = ComposeEnvService.refresh_for_project(app.root_path, app.compose_file)
    # Apply the regenerated override the same way a deploy does.
    up = DockerService.compose_up(app.root_path, detach=True,
                                  compose_file=app.compose_file)
    return {
        'success': True,
        'wrote': [written] if written else [],
        'reloaded': bool(up.get('success')),
        'warning': None if up.get('success') else up.get('error'),
    }


register_check({
    'type': 'compose_override',
    'title': 'Compose env overrides',
    'supported': _linux_only,
    'list_resources': _compose_list_resources,
    'render_expected': _compose_render_expected,
    'repair': _compose_repair,
})

# ---------------------------------------------------------------------------
# manifest reconciliation (#18)
#
# Unlike the file-based checks above, this one compares *config*, not files:
# for every manifest-managed app it asks ManifestApplyService for the
# (expected, observed) pair over the manifest-declared surface (port,
# healthcheck, env keys, volumes, domains) and reports drift when they differ.
# Because there is no file on disk to read, we supply a custom ``read_actual``
# that re-derives the observed side instead of touching the filesystem. The
# synthetic path key is ``manifest://app/<id>``.
# ---------------------------------------------------------------------------


def _manifest_list_resources():
    """Every Application that is governed by a stored manifest."""
    from app.models.application import Application
    from app.services.manifest_apply_service import ManifestApplyService

    out = []
    rows = Application.query_active().filter(Application.project_id.isnot(None)).all()
    for app in rows:
        try:
            if ManifestApplyService.resolved_for_app(app) is not None:
                out.append((app.id, app.name))
        except Exception as e:  # noqa: BLE001 — an unresolvable app is an error, not an absence
            out.append(_list_error_entry(
                'manifest', app.id, app.name,
                f'Could not resolve this app\'s manifest: {e}'))
    return out


def _manifest_render_expected(app_id):
    from app.models.application import Application
    from app.services.manifest_apply_service import ManifestApplyService

    app = Application.query_active().filter_by(id=app_id).first()
    if app is None:
        return {}
    resolved = ManifestApplyService.resolved_for_app(app)
    if resolved is None:
        return {}
    expected, _observed = ManifestApplyService.drift_pair(app, resolved)
    return {f'manifest://app/{app_id}': json.dumps(expected, sort_keys=True)}


def _manifest_read_actual(paths):
    from app.models.application import Application
    from app.services.manifest_apply_service import ManifestApplyService

    actual = {}
    for path in paths:
        try:
            app_id = int(path.rsplit('/', 1)[-1])
        except (ValueError, IndexError):
            actual[path] = None
            continue
        app = Application.query_active().filter_by(id=app_id).first()
        resolved = ManifestApplyService.resolved_for_app(app) if app else None
        if app is None or resolved is None:
            actual[path] = None
            continue
        _expected, observed = ManifestApplyService.drift_pair(app, resolved)
        actual[path] = json.dumps(observed, sort_keys=True)
    return actual


def _manifest_repair(app_id):
    from app.models.application import Application
    from app.services.manifest_apply_service import ManifestApplyService

    app = Application.query_active().filter_by(id=app_id).first()
    if app is None:
        return {'success': False, 'error': 'Service not found'}
    if not app.project_id:
        return {'success': False, 'error': 'This service is not managed by a manifest'}
    result = ManifestApplyService.apply_stored(app.project_id)
    ok = bool(result.get('success', False))
    return {
        'success': ok,
        'wrote': [f'app:{app_id}'],
        'reloaded': ok,
        'error': None if ok else result.get('error'),
    }


register_check({
    'type': 'manifest',
    'title': 'Manifest reconciliation',
    'list_resources': _manifest_list_resources,
    'render_expected': _manifest_render_expected,
    'read_actual': _manifest_read_actual,
    'repair': _manifest_repair,
})

# NOTE: no systemd unit check — see the module docstring. The panel-templated
# gunicorn units (PythonService.create_gunicorn_service) render from inputs
# that aren't persisted (workers/user/.env snapshot), so an expected render
# can't be reproduced faithfully; everything else under /etc/systemd/system is
# installer-owned.