documents/confluence.py - master - drbye/gag

documents/confluence.py
501 строка16 KB
Marusin Dmitry
v6.1: fix friend audit findings — security hardening, user persistence, dead code removal
29 июл 2026, 15:40
29 июл 2026, 15:40adb0438
"""
Confluence Client - Atlassian Confluence API client.

Provides space and page sync with children support, attachments, and recursive fetching.
"""

import asyncio
import os
import socket
import xml.etree.ElementTree as ET
import logging
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any

import httpx


class ConfluenceContentType(str, Enum):
    STORAGE = "confluence_storage"
    EDITOR_V2 = "editor_v2"
    ADF = "editor_adf"


@dataclass
class ConfluenceAttachment:
    """Confluence attachment model."""

    attachment_id: str
    title: str
    mime_type: str
    file_size: int
    download_link: Optional[str] = None
    web_link: Optional[str] = None
    created_at: Optional[str] = None
    version: int = 1


@dataclass
class ConfluencePage:
    """Confluence page model."""

    page_id: str
    title: str
    space_key: str
    content: str
    content_type: str = "confluence_storage"
    parent_id: Optional[str] = None
    version: int = 1
    created_at: Optional[str] = None
    updated_at: Optional[str] = None
    url: Optional[str] = None
    children: List["ConfluencePage"] = field(default_factory=list)
    attachments: List[ConfluenceAttachment] = field(default_factory=list)


class ConfluenceClient:
    # Confluence REST APIv2 base
    API_VERSION = "v2"
    # Legacy API for attachments (REST API v1)
    LEGACY_API = "rest/api"

    def __init__(
        self,
        url: Optional[str] = None,
        email: Optional[str] = None,
        api_token: Optional[str] = None,
    ):
        self.url = url or os.getenv("CONFLUENCE_URL", "")
        self.email = email or os.getenv("CONFLUENCE_EMAIL", "")
        self.api_token = api_token or os.getenv("CONFLUENCE_API_TOKEN", "")
        self.base_url = f"{self.url}/wiki/api/{self.API_VERSION}"
        self.legacy_url = f"{self.url}/{self.LEGACY_API}"

        self._headers = {
            "Authorization": f"Basic {self._get_auth()}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    def _get_auth(self) -> str:
        import base64

        creds = f"{self.email}:{self.api_token}"
        return base64.b64encode(creds.encode()).decode()

    def _get_client(self) -> httpx.AsyncClient:
        """Return a shared httpx AsyncClient (lazy init)."""
        if not hasattr(self, "_client") or self._client is None:
            self._client = httpx.AsyncClient(
                headers=self._headers,
                timeout=60.0,
                follow_redirects=False,
            )
        return self._client

    # ──────────────────────────────────────────────────────────
    # Attachments API
    # ──────────────────────────────────────────────────────────

    async def list_attachments(
        self,
        page_id: str,
    ) -> List[Dict[str, Any]]:
        """List all attachments for a page."""
        client = self._get_client()
        resp = await client.get(
            f"{self.legacy_url}/content/{page_id}/child/attachment",
            headers=self._headers,
            params={"limit": 100},
            timeout=30.0,
        )
        if resp.status_code != 200:
            return []
        data = resp.json()
        return data.get("results", [])

    async def get_attachment(
        self,
        page_id: str,
        attachment_id: str,
    ) -> Optional[Dict[str, Any]]:
        """Get a specific attachment."""
        client = self._get_client()
        resp = await client.get(
            f"{self.legacy_url}/content/{page_id}/child/attachment/{attachment_id}",
            headers=self._headers,
            timeout=30.0,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

    async def download_attachment(
        self,
        page_id: str,
        attachment_id: str,
    ) -> Optional[bytes]:
        """Download attachment binary content."""
        attachment = await self.get_attachment(page_id, attachment_id)
        if not attachment:
            return None

        download_url = attachment.get("_links", {}).get("download")
        if not download_url:
            return None

        # Handle relative URLs
        if download_url.startswith("/"):
            download_url = f"{self.url}{download_url}"

        # SSRF protection: resolve hostname once, validate, then use literal IP
        # to prevent TOCTOU DNS rebinding attacks.
        try:
            parsed = httpx.URL(download_url)
            hostname = parsed.host
            path = parsed.full_path or "/"
            if not hostname:
                raise ValueError("Missing hostname in download URL")

            # Same-domain check: resolve via our own getaddrinfo (same call
            # used by _is_private_ip), avoiding httpx's second resolution.
            is_private, resolved_ip = self._is_private_ip(hostname)
            expected_domain = self.url.replace("https://", "").replace("http://", "").split("/")[0]
            if hostname == expected_domain:
                pass  # Same domain, use hostname as-is (trusted)
                final_url = download_url
                final_headers = dict(self._headers)
            elif is_private or not resolved_ip:
                raise ValueError("Private IP addresses not allowed or unresolvable hostname")
            else:
                # Use literal IP with Host header to eliminate TOCTOU window
                final_url = f"https://{resolved_ip}{path}"
                final_headers = {**self._headers, "Host": hostname}
        except ValueError:
            raise
        except Exception as e:
            logging.getLogger(__name__).warning(f"SSRF validation failed for {download_url}: {e}")
            return None

        client = self._get_client()
        resp = await client.get(
            final_url,
            headers=final_headers,
            timeout=60.0,
        )
        if resp.status_code == 200:
            return resp.content
        return None

    def _is_private_ip(self, hostname: str) -> tuple[bool, str]:
        """Check if hostname resolves to any private IP address.

        SECURITY: Must check ALL resolved IPs, not just the first.
        A hostname that resolves to both public and private IPs enables
        DNS rebinding attacks — the first lookup returns public (passes
        the check), then the actual request goes to the private IP.

        Returns:
            (is_private, first_resolved_ip) — the IP is empty string
            on resolution failure.
        """
        import ipaddress
        try:
            addrs = socket.getaddrinfo(hostname, None)
            first_ip = ""
            for addr in addrs:
                ip = addr[4][0]
                if not first_ip:
                    first_ip = ip
                network = ipaddress.ip_address(ip)
                if network.is_private or network.is_loopback or network.is_link_local:
                    return True, first_ip
            return False, first_ip
        except Exception:
            pass
        return False, ""

    async def get_page_attachments(
        self,
        page_id: str,
    ) -> List[ConfluenceAttachment]:
        """Get all attachments for a page with full metadata."""
        attachments = []
        results = await self.list_attachments(page_id)

        for item in results:
            file_meta = item.get("metadata", {}).get("file", {})
            attachments.append(
                ConfluenceAttachment(
                    attachment_id=item.get("id", ""),
                    title=item.get("title", ""),
                    mime_type=file_meta.get("mimeType", "application/octet-stream"),
                    file_size=file_meta.get("size", 0),
                    download_link=item.get("_links", {}).get("download"),
                    web_link=item.get("_links", {}).get("webui"),
                    created_at=item.get("createdAt"),
                    version=item.get("version", {}).get("number", 1),
                )
            )

        return attachments

    # ──────────────────────────────────────────────────────────
    # Child pages with recursive support
    # ──────────────────────────────────────────────────────────

    async def get_page_tree(
        self,
        page_id: str,
        include_attachments: bool = False,
    ) -> ConfluencePage:
        """Get a page with its full tree of children and optionally attachments."""
        page_data = await self.get_page(page_id)
        if not page_data:
            return ConfluencePage(
                page_id=page_id,
                title="Not Found",
                space_key="",
                content="",
            )

        body = await self.get_page_body(page_id)

        page = ConfluencePage(
            page_id=page_id,
            title=page_data.get("title", ""),
            space_key=page_data.get("spaceId", ""),
            content=body or "",
            version=page_data.get("version", {}).get("number", 1),
            created_at=page_data.get("createdAt"),
            updated_at=page_data.get("version", {}).get("createdAt"),
            url=page_data.get("_links", {}).get("webui"),
        )

        # Get children recursively
        page.children = await self.get_page_children(page_id, depth=10)

        if include_attachments:
            page.attachments = await self.get_page_attachments(page_id)

        return page

    async def sync_pages_with_tree(
        self,
        page_ids: List[str],
        include_attachments: bool = True,
    ) -> List[ConfluencePage]:
        """Sync multiple pages with full tree and attachments."""
        pages = []

        for page_id in page_ids:
            page = await self.get_page_tree(page_id, include_attachments)
            pages.append(page)

        return pages

    # ──────────────────────────────────────────────────────────
    # Legacy API wrappers (keep for compatibility)
    # ──────────────────────────────────────────────────────────

    async def list_spaces(self, limit: int = 25) -> List[Dict[str, Any]]:
        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/spaces",
            headers=self._headers,
            params={"limit": limit},
            timeout=30.0,
        )
        resp.raise_for_status()
        data = resp.json()
        return data.get("results", [])

    async def get_space(self, space_key: str) -> Optional[Dict[str, Any]]:
        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/spaces/{space_key}",
            headers=self._headers,
            timeout=30.0,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

    async def list_pages(
        self,
        space_key: Optional[str] = None,
        limit: int = 25,
    ) -> List[Dict[str, Any]]:
        params = {"limit": limit}
        if space_key:
            params["space-id"] = space_key

        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/pages",
            headers=self._headers,
            params=params,
            timeout=30.0,
        )
        resp.raise_for_status()
        data = resp.json()
        return data.get("results", [])

    async def get_page(self, page_id: str) -> Optional[Dict[str, Any]]:
        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/pages/{page_id}",
            headers=self._headers,
            timeout=30.0,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

    async def get_page_body(self, page_id: str) -> Optional[str]:
        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/pages/{page_id}/body",
            headers=self._headers,
            timeout=30.0,
        )
        if resp.status_code == 200:
            data = resp.json()
            return data.get("representation", {}).get("value", "")
        return None

    async def get_page_children(
        self,
        page_id: str,
        depth: int = 1,
    ) -> List[ConfluencePage]:
        children = []

        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/pages/{page_id}/children",
            headers=self._headers,
            timeout=30.0,
        )
        if resp.status_code != 200:
            return children

        data = resp.json()
        results = data.get("results", [])

        # Fetch all child pages and bodies in parallel to avoid N+1
        async def _fetch_child(item: dict) -> Optional[ConfluencePage]:
            child_id = item.get("id")
            child_page, body = await asyncio.gather(
                self.get_page(child_id),
                self.get_page_body(child_id),
            )
            if not child_page:
                return None
            child = ConfluencePage(
                page_id=child_id,
                title=child_page.get("title", ""),
                space_key=child_page.get("spaceId", ""),
                content=body or "",
                version=child_page.get("version", {}).get("number", 1),
            )
            if depth > 1:
                grandchildren = await self.get_page_children(
                    child_id, depth - 1
                )
                child.children = grandchildren
            return child

        child_results = await asyncio.gather(
            *[_fetch_child(item) for item in results],
            return_exceptions=True,
        )
        for cr in child_results:
            if isinstance(cr, ConfluencePage):
                children.append(cr)

        return children

    async def sync_space(
        self,
        space_key: str,
        include_children: bool = True,
        max_depth: int = 3,
    ) -> List[ConfluencePage]:
        pages = []

        space_pages = await self.list_pages(space_key=space_key, limit=100)

        for page_data in space_pages:
            page_id = page_data.get("id")
            body = await self.get_page_body(page_id)

            page = ConfluencePage(
                page_id=page_id,
                title=page_data.get("title", ""),
                space_key=space_key,
                content=body or "",
                version=page_data.get("version", {}).get("number", 1),
            )

            if include_children and max_depth > 0:
                page.children = await self.get_page_children(page_id, depth=max_depth)

            pages.append(page)

        return pages

    async def sync_pages(
        self,
        page_ids: List[str],
        include_children: bool = False,
    ) -> List[ConfluencePage]:
        pages = []

        for page_id in page_ids:
            page_data = await self.get_page(page_id)
            if not page_data:
                continue

            body = await self.get_page_body(page_id)

            page = ConfluencePage(
                page_id=page_id,
                title=page_data.get("title", ""),
                space_key=page_data.get("spaceId", ""),
                content=body or "",
            )

            if include_children:
                page.children = await self.get_page_children(page_id)

            pages.append(page)

        return pages

    async def search_pages(
        self,
        cql: str,
        limit: int = 25,
    ) -> List[Dict[str, Any]]:
        client = self._get_client()
        resp = await client.get(
            f"{self.base_url}/pages",
            headers=self._headers,
            params={"cql": cql, "limit": limit},
            timeout=30.0,
        )
        resp.raise_for_status()
        data = resp.json()
        return data.get("results", [])


_confluence_client: Optional[ConfluenceClient] = None


def get_confluence_client() -> ConfluenceClient:
    global _confluence_client
    if _confluence_client is None:
        _confluence_client = ConfluenceClient()
    return _confluence_client