vulncheck/app/services/m365_service.py

"""
Microsoft 365 Apps CVE detection (Plan P).

Microsoft 365 Apps (formerly Office 365 ProPlus) security fixes are NOT
published to NVD and are NOT detected by Wazuh's vulnerability detector —
they only live on one human-readable Microsoft Learn page:

    https://learn.microsoft.com/en-us/officeupdates/microsoft365-apps-security-updates

There is no Microsoft API. So we parse that page, learn the latest patched
build per update channel, compare it against the build Wazuh's syscollector
reports as installed, and create real-CVE vulnerability rows for every
monthly update the host is behind on.

Build logic (verified against the tester's example):
  installed 16.0.19929.20172  vs  Monthly Enterprise Channel 19929.20162
  -> 20172 >= 20162  -> UNAFFECTED  (no CVEs)
  installed < a section's channel build -> AFFECTED -> attach that
  section's CVEs (union across every section the host is behind on).

Channel mapping (tester's rule): the deployed channel isn't in the
syscollector name, so we approximate it from the product name —
"...enterprise..." -> Monthly Enterprise Channel, else Current Channel.
"""
import json
import logging
import re
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple

import httpx
import lxml.html
from sqlalchemy.orm import Session

from app.models.setting import Setting

logger = logging.getLogger(__name__)

M365_SECURITY_URL = (
    "https://learn.microsoft.com/en-us/officeupdates/"
    "microsoft365-apps-security-updates"
)

# ---------- cache (settings table) ----------
M365_CACHE_KEY = "m365_security_cache"
M365_CACHE_TS_KEY = "m365_security_cache_updated_at"
M365_TTL_HOURS = 24

# ---------- toggle ----------
SETTING_M365_ENABLED = "m365_detection_enabled"

HTTP_TIMEOUT = 30.0

# Build line: "Monthly Enterprise Channel: Version 2604 (Build 19929.20162)"
_BUILD_RE = re.compile(
    r"([A-Za-z0-9()/ \-]+?):\s*Version\s+(\d{3,4})\s*\(\s*Build\s+(\d+\.\d+)\s*\)"
)
# Month-day-year heading that delimits each monthly section.
_DATE_RE = re.compile(
    r"\b(January|February|March|April|May|June|July|August|September|"
    r"October|November|December)\s+(\d{1,2}),\s+(\d{4})\b"
)
_CVE_RE = re.compile(r"CVE-\d{4}-\d{4,}", re.IGNORECASE)

# Product-name -> channel approximation.
CHANNEL_MONTHLY_ENTERPRISE = "Monthly Enterprise Channel"
CHANNEL_CURRENT = "Current Channel"


class M365Error(Exception):
    """Raised when the M365 security page cannot be fetched/parsed."""


# ============================================================
# build helpers
# ============================================================

def parse_build(version: str) -> Optional[Tuple[int, int]]:
    """'16.0.19929.20172' or '19929.20172' -> (19929, 20172).

    Microsoft 365 build numbers are the last two dotted segments
    (BBBBB.RRRRR). The leading '16.0.' is the Office major and is
    constant, so we ignore it.
    """
    if not version:
        return None
    nums = re.findall(r"\d+", version)
    if len(nums) < 2:
        return None
    try:
        return int(nums[-2]), int(nums[-1])
    except ValueError:
        return None


def channel_for_product(product_name: str) -> str:
    """Tester's rule: name contains 'enterprise' -> MEC, else Current."""
    return (
        CHANNEL_MONTHLY_ENTERPRISE
        if "enterprise" in (product_name or "").lower()
        else CHANNEL_CURRENT
    )


def is_m365_apps(product_name: str) -> bool:
    """True for syscollector entries like 'Microsoft 365 Apps for enterprise'."""
    n = (product_name or "").lower()
    return "microsoft 365 apps" in n or "office 365 proplus" in n


# ============================================================
# page fetch + parse
# ============================================================

def _parse_security_page(html: str) -> List[dict]:
    """Parse the MS365 security page into a list of monthly releases.

    Each release: {
        "date": "May 12, 2026",
        "channel_builds": {channel_name: [(major, rev), ...]},  # max = newest
        "cves": ["CVE-2026-40361", ...],   # every CVE in the section
    }
    Releases are returned in page order (newest first).
    """
    # Flatten to text in document order. The page is a linear sequence of
    # date headings -> channel/build lines -> product headings -> CVE
    # bullets, so segmenting the flattened text by date heading is robust
    # against markup churn.
    doc = lxml.html.fromstring(html)
    for bad in doc.xpath("//script | //style | //nav | //header | //footer"):
        bad.getparent().remove(bad)
    body = doc.xpath("//main") or [doc]
    text = body[0].text_content()

    # Find date-heading anchors and slice between them.
    matches = list(_DATE_RE.finditer(text))
    releases: List[dict] = []
    for i, m in enumerate(matches):
        start = m.end()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
        section = text[start:end]
        date_label = f"{m.group(1)} {m.group(2)}, {m.group(3)}"

        channel_builds: Dict[str, List[Tuple[int, int]]] = {}
        for bm in _BUILD_RE.finditer(section):
            channel = bm.group(1).strip()
            build = parse_build(bm.group(3))
            if build:
                channel_builds.setdefault(channel, []).append(build)
        if not channel_builds:
            # Not a real release section (e.g. intro paragraph mentioning
            # a date) — skip.
            continue

        cves = sorted({c.upper() for c in _CVE_RE.findall(section)})
        if not cves:
            continue

        releases.append({
            "date": date_label,
            "channel_builds": channel_builds,
            "cves": cves,
        })
    return releases


def _load_cache(db: Session) -> Optional[List[dict]]:
    ts = db.query(Setting).filter(Setting.key == M365_CACHE_TS_KEY).first()
    cache = db.query(Setting).filter(Setting.key == M365_CACHE_KEY).first()
    if not ts or not cache or not cache.value:
        return None
    try:
        if datetime.now() - datetime.fromisoformat(ts.value) > timedelta(hours=M365_TTL_HOURS):
            return None
        return json.loads(cache.value)
    except (ValueError, json.JSONDecodeError):
        return None


def _store_cache(db: Session, releases: List[dict]) -> None:
    s = db.query(Setting).filter(Setting.key == M365_CACHE_KEY).first()
    if s:
        s.value = json.dumps(releases)
    else:
        db.add(Setting(key=M365_CACHE_KEY, value=json.dumps(releases),
                       description="MS365 Apps security-updates parse cache (24h)"))
    ts = db.query(Setting).filter(Setting.key == M365_CACHE_TS_KEY).first()
    if ts:
        ts.value = datetime.now().isoformat()
    else:
        db.add(Setting(key=M365_CACHE_TS_KEY, value=datetime.now().isoformat(),
                       description="Timestamp of last MS365 page parse"))
    db.commit()


def fetch_security_data(db: Session, force_refresh: bool = False) -> List[dict]:
    """Return parsed monthly releases, cached 24h in the settings table."""
    if not force_refresh:
        cached = _load_cache(db)
        if cached is not None:
            return cached
    try:
        with httpx.Client(timeout=HTTP_TIMEOUT, follow_redirects=True,
                          headers={"User-Agent": "VulnCheck/1.0"}) as client:
            resp = client.get(M365_SECURITY_URL)
            resp.raise_for_status()
            html = resp.text
    except httpx.HTTPError as e:
        raise M365Error(f"could not fetch MS365 security page: {e}") from e

    releases = _parse_security_page(html)
    if not releases:
        raise M365Error("MS365 page parsed to zero releases — layout changed?")
    _store_cache(db, releases)
    logger.info("MS365: parsed %d monthly releases", len(releases))
    return releases


# ============================================================
# detection
# ============================================================

def _channel_max(release: dict, channel: str) -> Optional[Tuple[int, int]]:
    """Newest (max) build for `channel` in a release, as a tuple."""
    builds = release.get("channel_builds", {}).get(channel)
    if not builds:
        return None
    # builds may be lists from JSON -> normalise to tuples
    return max(tuple(b) for b in builds)


def detect_missing_cves(
    releases: List[dict],
    *,
    installed_version: str,
    channel: str,
) -> dict:
    """Compare an installed M365 build against the parsed releases.

    Returns {
        "affected": bool,
        "installed_build": "19929.20172" or None,
        "latest_build": "19929.20162" or None,   # newest patched, this channel
        "missing_cves": [ ... ],                  # union, deduped
        "behind_releases": [ "May 12, 2026", ... ],
    }
    """
    out = {
        "affected": False,
        "installed_build": None,
        "latest_build": None,
        "missing_cves": [],
        "behind_releases": [],
    }
    installed = parse_build(installed_version)
    if not installed:
        return out
    out["installed_build"] = f"{installed[0]}.{installed[1]}"

    # Newest patched build for this channel across the whole page.
    channel_builds = [b for r in releases if (b := _channel_max(r, channel))]
    if not channel_builds:
        return out
    latest = max(channel_builds)
    out["latest_build"] = f"{latest[0]}.{latest[1]}"

    if installed >= latest:
        return out  # fully patched -> unaffected

    # Behind: union CVEs from every section whose channel build the host
    # has not reached.
    out["affected"] = True
    cve_set: set = set()
    for r in releases:
        b = _channel_max(r, channel)
        if b and installed < b:
            cve_set.update(r.get("cves", []))
            out["behind_releases"].append(r.get("date"))
    out["missing_cves"] = sorted(cve_set)
    return out


def upsert_m365_vulnerability(
    db: Session,
    *,
    asset_id: int,
    cve_id: str,
    product_name: str,
    installed_version: str,
    fixed_build: Optional[str],
) -> Tuple[Optional[int], bool]:
    """Create/refresh a real-CVE M365 vuln row. Returns (id, was_created).

    CVSS/severity are left as a neutral placeholder; the nightly
    enrichment (EPSS/KEV/NVD dates) and the Correct-CVSS job refine them.
    These are real CVE ids, so they enrich like any other CVE.
    """
    from app.models.vulnerability import (
        Vulnerability, VulnerabilitySeverity, VulnerabilityStatus,
    )

    cve_id = cve_id.upper()
    existing = (
        db.query(Vulnerability)
        .filter(Vulnerability.cve_id == cve_id, Vulnerability.asset_id == asset_id)
        .first()
    )
    title = f"{product_name} — {cve_id} (Microsoft 365 Apps security update)"
    desc = (
        f"{cve_id} is fixed by a Microsoft 365 Apps security update not yet "
        f"applied on this host. Installed build {installed_version}; patched "
        f"in build {fixed_build or 'unknown'} or later. Source: Microsoft 365 "
        f"Apps security updates page (not published to NVD / not seen by Wazuh)."
    )

    if existing:
        existing.title = title[:500]
        existing.description = desc
        existing.package_name = product_name[:255]
        existing.package_version = installed_version[:100]
        existing.fixed_version = (fixed_build or None)
        if existing.status == VulnerabilityStatus.patched:
            existing.status = VulnerabilityStatus.open
            existing.patched_at = None
        existing.detected_at = datetime.now()
        try:
            existing.refresh_scores()
        except Exception:
            pass
        return existing.id, False

    vuln = Vulnerability(
        cve_id=cve_id,
        asset_id=asset_id,
        cvss_score=None,
        severity=VulnerabilitySeverity.medium,  # placeholder; enrichment refines
        status=VulnerabilityStatus.open,
        title=title[:500],
        description=desc,
        package_name=product_name[:255],
        package_version=installed_version[:100],
        fixed_version=(fixed_build or None),
        detected_at=datetime.now(),
        sources='["microsoft365-apps"]',
        first_detected_by="m365_check",
    )
    db.add(vuln)
    db.flush()
    try:
        vuln.refresh_scores()
    except Exception:
        pass
    return vuln.id, True


# ============================================================
# orchestration (shared by the endpoint and the nightly job)
# ============================================================

def run_m365_check(db: Session, wazuh, asset_id: Optional[int] = None) -> dict:
    """Walk Wazuh-linked assets, detect M365-Apps CVE exposure, upsert rows.

    `wazuh` is an already-configured WazuhClient (the caller owns its
    lifecycle, matching the eol-check pattern).
    """
    from app.models.asset import Asset

    releases = fetch_security_data(db)

    q = db.query(Asset).filter(Asset.wazuh_agent_id.isnot(None))
    if asset_id is not None:
        q = q.filter(Asset.id == asset_id)
    assets = q.all()

    stats = {
        "assets_scanned": 0,
        "m365_installs": 0,
        "assets_affected": 0,
        "cve_findings_total": 0,
        "cve_findings_new": 0,
        "releases_parsed": len(releases),
        "errors": [],
    }

    for asset in assets:
        try:
            pkgs = wazuh.get_packages(asset.wazuh_agent_id) or []
        except Exception as e:
            stats["errors"].append(f"asset {asset.id} ({asset.hostname}): {e}")
            continue
        stats["assets_scanned"] += 1

        # An asset can list the same product per language pack (de-de,
        # en-us, .proof, ...) — collapse to one detection per build.
        seen_builds: set = set()
        asset_affected = False
        for pkg in pkgs:
            name = (pkg.get("name") or "").strip()
            version = (pkg.get("version") or "").strip()
            if not name or not version or not is_m365_apps(name):
                continue
            stats["m365_installs"] += 1
            channel = channel_for_product(name)
            key = (channel, version)
            if key in seen_builds:
                continue
            seen_builds.add(key)

            result = detect_missing_cves(
                releases, installed_version=version, channel=channel
            )
            if not result["affected"]:
                continue
            asset_affected = True
            for cve_id in result["missing_cves"]:
                try:
                    _, created = upsert_m365_vulnerability(
                        db,
                        asset_id=asset.id,
                        cve_id=cve_id,
                        product_name=name,
                        installed_version=version,
                        fixed_build=result["latest_build"],
                    )
                    stats["cve_findings_total"] += 1
                    if created:
                        stats["cve_findings_new"] += 1
                except Exception as e:
                    logger.warning(
                        "M365 upsert failed (%s on asset %s): %s",
                        cve_id, asset.id, e,
                    )
        if asset_affected:
            stats["assets_affected"] += 1

    db.commit()
    logger.info(
        "M365 check: %d assets scanned, %d installs, %d affected, "
        "%d CVE rows (%d new)",
        stats["assets_scanned"], stats["m365_installs"],
        stats["assets_affected"], stats["cve_findings_total"],
        stats["cve_findings_new"],
    )
    return stats