10 · 运维中心与清理加速：闭环、安全边界与"宁可少删不可误删"

"""
ops/incidents.py — 故障事件管理：告警自动聚合为 Incident
"""
from datetime import timedelta

from django.utils import timezone

LEVEL_ORDER = {'P0': 0, 'P1': 1, 'P2': 2, '提示': 3}


def attach_or_create(event):
    """把一条新触发的告警并入现有故障单（30 分钟窗口）或新建故障单"""
    from .models import Incident

    if event.policy_id:
        recent = Incident.objects.filter(
            status='open',
            started_at__gte=timezone.now() - timedelta(minutes=30),
        ).order_by('-started_at').first()
        if recent:
            event.incident = recent
            event.save(update_fields=['incident'])
            # 级别取更高者（P0 < P1 < P2 < 提示 的字典序即严重序）
            if LEVEL_ORDER.get(event.level, 9) < LEVEL_ORDER.get(recent.level, 9):
                Incident.objects.filter(pk=recent.pk).update(level=event.level)
            # 并入时补全涉及服务（首次创建只取了首条策略的域）
            svc = event.policy.metric_key.split('.')[0] if event.policy else ''
            if svc and svc not in recent.services.split(','):
                new_services = f'{recent.services},{svc}'.strip(',')[:200]
                Incident.objects.filter(pk=recent.pk).update(services=new_services)
            return recent
    incident = Incident.objects.create(
        title=event.summary[:150],
        level=event.level,
        started_at=event.started_at,
        services=event.policy.metric_key.split('.')[0] if event.policy else '',
    )
    event.incident = incident
    event.save(update_fields=['incident'])
    return incident


def maybe_resolve(incident):
    """故障单下所有告警恢复后，自动标记恢复"""
    firing = incident.alerts.filter(status='firing').count()
    if firing == 0 and incident.status == 'open':
        incident.status = 'resolved'
        incident.resolved_at = timezone.now()
        incident.save(update_fields=['status', 'resolved_at'])
        return True
    return False

    from ops.incidents import attach_or_create, maybe_resolve  # 取消注释
    ...
    attach_or_create(event)          # 并入/创建故障单（取消注释）
    process_heal_for_event(event)    # 自愈动作（取消注释）

"""
ops/urlsafe.py — 出站 URL 统一安全校验（拨测 / 自愈回调 / 通知 Webhook 共用）

规则：
- 仅允许 http/https 且必须带主机名；
- 解析域名后逐 IP 检查：链路本地（含 169.254.169.254 云元数据）、保留、
  组播地址始终阻断；环回与私网地址默认允许（自用/内网拨测的合法目标），
  可分别用环境变量关闭：OBS_ALLOW_LOOPBACK_URL=0 / OBS_ALLOW_PRIVATE_URL=0；
- 所有出站请求必须经 open_no_redirect() 发出：禁止 30x 自动重定向
  （否则攻击者用外网跳板 302 到内网即可绕过全部 IP 黑名单）。

已知边界：校验与请求之间存在 DNS rebinding TOCTOU 窗口（校验时解析一次、
连接时再解析一次）。内网部署如需彻底封堵，可固定解析结果直连 IP。
"""
import ipaddress
import os
import socket
import urllib.error
import urllib.request
from urllib.parse import urlsplit

_ALLOW_LOOPBACK = os.environ.get('OBS_ALLOW_LOOPBACK_URL', '1') == '1'
_ALLOW_PRIVATE = os.environ.get('OBS_ALLOW_PRIVATE_URL', '1') == '1'


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """30x 一律不跟随：redirect_request 返回 None 会让 urlopen 抛 HTTPError"""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


_OPENER = urllib.request.build_opener(_NoRedirect)


def check_ip(ip):
    """单 IP 分类检查：返回错误信息或 None"""
    ip = ipaddress.ip_address(ip)
    if ip.is_link_local or ip.is_reserved or ip.is_multicast:
        return f'目标地址被禁止: {ip}（链路本地/保留/组播地址）'
    if ip.is_loopback and not _ALLOW_LOOPBACK:
        return f'目标为环回地址 {ip}，且已禁用环回 URL'
    if ip.is_private and not _ALLOW_PRIVATE:
        return f'目标为私网地址 {ip}，且已禁用私网 URL'
    return None


def validate_url(url):
    """出站 URL 校验：返回错误信息或 None"""
    parts = urlsplit(url or '')
    if parts.scheme not in ('http', 'https'):
        return 'URL 仅允许 http/https'
    if not parts.hostname:
        return 'URL 缺少主机名'
    if parts.username or parts.password:
        return 'URL 不应包含用户名/密码'
    try:
        infos = socket.getaddrinfo(parts.hostname, None)
    except OSError as e:
        return f'域名解析失败: {e}'
    for info in infos:
        err = check_ip(info[4][0])
        if err:
            return err
    return None


def open_no_redirect(url, timeout=10, headers=None):
    """校验 URL 并以"禁止重定向"的 opener 发起 GET。

    返回 (response, None)；校验失败返回 (None, 错误信息)；
    3xx 重定向会作为 HTTPError 抛出（不跟随），由调用方按失败处理。
    """
    err = validate_url(url)
    if err:
        return None, err
    req = urllib.request.Request(url, headers=headers or {})
    try:
        resp = _OPENER.open(req, timeout=timeout)
        return resp, None
    except urllib.error.HTTPError:
        raise  # 含 3xx：让调用方与 4xx/5xx 同样按失败处理
    except OSError as e:
        return None, f'请求失败: {e}'

# 可执行文件白名单（绝对路径）：未配置时 command 类型禁用。
# 例：Windows  set OBS_HEAL_CMD_ALLOWLIST=C:\ops\clean.exe;C:\ops\restartsvc.exe
#      Linux   export OBS_HEAL_CMD_ALLOWLIST=/usr/local/bin/clean-tmp.sh:/usr/bin/systemctl
_CMD_ALLOWLIST = [
    str(Path(p).resolve())
    for p in (__import__('os').environ.get('OBS_HEAL_CMD_ALLOWLIST') or '').split(
        ';' if sys.platform == 'win32' else ':')
    if p.strip()
]

def process_heal_for_event(event):
    """告警事件触发时调用：找该策略绑定的启用动作，冷却期外执行"""
    actions = HealAction.objects.filter(policy=event.policy, enabled=True)
    for action in actions:
        # 冷却期原子占位：条件 UPDATE 抢占，抢到的才执行
        deadline = timezone.now() - timedelta(seconds=action.cooldown_sec)
        claimed = HealRun.objects.filter(
            action=action, started_at__gte=deadline,
        ).exists() is False
        if not claimed:
            continue
        run = run_action(action, reason=f'告警：{event.summary}')
        ...

# ---------------- 演示模式（只读展示账号） ----------------
DEMO_GROUP_NAME = '演示访客'

# 演示账号禁止访问的路径前缀：管理入口 / 含令牌或密钥的页面 / 会写数据的页面
DEMO_BLOCKED_PREFIXES = (
    '/ops/notify/',      # 通知渠道配置（含 SMTP 密码）
    '/ops/audit/',       # 操作审计
    '/cleaner/',         # 清理加速（高危维护操作）
    '/ops/heal/',        # 自愈动作管理
    '/ops/probe/',       # 拨测管理（任务 URL 可能含接入令牌）
    '/alerts/policies/', # 告警策略管理
    '/integration/',     # 接入中心（含接入令牌明文）
    '/forum/chaos/',     # 故障演练（会写数据）
    '/metrics',          # Prometheus 全量指标
    '/admin/',           # 后台
)

LOGOUT_PATH = '/accounts/logout/'


def is_demo_user(user):
    """判断是否演示账号：属于"演示访客"组即可（结果缓存在 user 对象上）"""
    if user is None or not getattr(user, 'is_authenticated', False):
        return False
    cached = getattr(user, '_obs_is_demo', None)
    if cached is not None:
        return cached
    try:
        result = user.groups.filter(name=DEMO_GROUP_NAME).exists()
    except Exception:
        # 判定失败时按"非演示"放行会绕过写保护，这里保守地按演示账号处理（fail-closed）
        return True
    try:
        user._obs_is_demo = result
    except AttributeError:
        pass
    return result


def _demo_403(request):
    from django.shortcuts import render
    return render(request, 'demo_403.html', status=403)

        user = getattr(request, 'user', None)
        if user is not None and user.is_authenticated:
            # ---- 演示模式：只读展示账号 ----
            if is_demo_user(user):
                # 写保护：除退出登录外，任何非 GET 请求一律 403
                if request.method not in ('GET', 'HEAD', 'OPTIONS') \
                        and path != LOGOUT_PATH:
                    return _demo_403(request)
                # 敏感页面拦截：管理入口 / 含令牌密钥的页面
                if any(path.startswith(p) for p in DEMO_BLOCKED_PREFIXES):
                    return _demo_403(request)
                return self.get_response(request)
            return self.get_response(request)

        from ops.workers import probe_loop, inspect_loop

        threads = (
            ('obs-host-collector', host_collector_loop, 'host-collector'),
            ('obs-alert-engine', alert_engine_loop, 'alert-engine'),
            ('obs-probe', probe_loop, 'probe'),
            ('obs-inspect', inspect_loop, 'inspect'),
        )