import logging import os from email.utils import parsedate_to_datetime from httpx import ( AsyncBaseTransport, AsyncClient, Auth, BasicAuth, Request, Response, Timeout, ) from ..controllers.notes_search import NotesSearchController from ..http import nextcloud_httpx_transport from .calendar import CalendarClient from .collectives import CollectivesClient from .contacts import ContactsClient from .cookbook import CookbookClient from .deck import DeckClient from .groups import GroupsClient from .mail import MailClient from .news import NewsClient from .notes import NotesClient from .sharing import SharingClient from .tables import TablesClient from .talk import TalkClient from .users import UsersClient from .webdav import WebDAVClient from .webhooks import WebhooksClient logger = logging.getLogger(__name__) async def log_request(request: Request): logger.debug( "Request event hook: %s %s - Waiting for content", request.method, request.url, ) logger.debug("Request body: %s", request.content) logger.debug("Headers: %s", request.headers) async def log_response(response: Response): await response.aread() logger.debug("Response [%s] %s", response.status_code, response.text) def _normalise_search_result(item: dict) -> dict: """Normalise a webdav.search_files item to the get_files_by_tag shape.""" path = item.get("path", "") if path and not path.startswith("/"): path = "/" + path last_modified_timestamp = item.get("last_modified_timestamp") last_modified = item.get("last_modified") if last_modified_timestamp is None and last_modified: try: last_modified_timestamp = int( parsedate_to_datetime(last_modified).timestamp() ) except (TypeError, ValueError): last_modified_timestamp = None file_id = item.get("file_id") if item.get("file_id") is not None else item.get("id") return { "id": file_id, "path": path, "name": item.get("name") or (path.rsplit("/", 1)[-1] if path else ""), "size": item.get("size", 0), "content_type": item.get("content_type", ""), "last_modified": last_modified, "last_modified_timestamp": last_modified_timestamp, "etag": item.get("etag"), "is_directory": item.get("is_directory", False), } class AsyncDisableCookieTransport(AsyncBaseTransport): """This Transport disable cookies from accumulating in the httpx AsyncClient Thanks to: https://github.com/encode/httpx/issues/2992#issuecomment-2133258994 """ def __init__(self, transport: AsyncBaseTransport): self.transport = transport async def handle_async_request(self, request: Request) -> Response: response = await self.transport.handle_async_request(request) response.headers.pop("set-cookie", None) return response class NextcloudClient: """Main Nextcloud client that orchestrates all app clients.""" def __init__( self, base_url: str, username: str, auth: Auth | None = None, *, auth_username: str | None = None, password: str | None = None, token: str | None = None, ): # ``username`` is the Nextcloud UID — it drives DAV/API path # construction (e.g. ``/remote.php/dav/files//``). ``auth_username`` # is the credential identity Nextcloud authenticates the app password # against (the loginName), which differs from the UID for # OIDC-provisioned users. Defaults to ``username`` so single-user and # OAuth modes (where UID == loginName) are unchanged. Callers pass the # matching ``auth=BasicAuth(auth_username, ...)`` for the httpx leg; # ``auth_username`` is threaded to the CalDAV client, which builds its # own auth object from the raw credential. self.username = username auth_username = auth_username or username self._client = AsyncClient( base_url=base_url, auth=auth, transport=AsyncDisableCookieTransport(nextcloud_httpx_transport()), event_hooks={"request": [log_request], "response": [log_response]}, timeout=Timeout(timeout=30, connect=5), ) # Initialize app clients self.notes = NotesClient(self._client, username) self.webdav = WebDAVClient(self._client, username) self.tables = TablesClient(self._client, username) # CalendarClient takes raw credentials so caldav (which uses niquests as # its preferred backend in v3.x) builds a backend-compatible auth object # itself — passing httpx.BasicAuth here breaks under niquests (#731). self.calendar = CalendarClient( base_url, username, auth_username=auth_username, password=password, token=token, ) self.contacts = ContactsClient(self._client, username) self.cookbook = CookbookClient(self._client, username) self.collectives = CollectivesClient(self._client, username) self.deck = DeckClient(self._client, username) self.news = NewsClient(self._client, username) self.mail = MailClient(self._client, username) self.talk = TalkClient(self._client, username) self.users = UsersClient(self._client, username) self.groups = GroupsClient(self._client, username) self.sharing = SharingClient(self._client, username) self.webhooks = WebhooksClient(self._client, username) # Initialize controllers self._notes_search = NotesSearchController() @classmethod def from_env(cls): logger.info("Creating NC Client using env vars") host = os.environ["NEXTCLOUD_HOST"] username = os.environ["NEXTCLOUD_USERNAME"] password = os.environ["NEXTCLOUD_PASSWORD"] # Pass username to constructor return cls( base_url=host, username=username, auth=BasicAuth(username, password), password=password, ) @classmethod def from_token(cls, base_url: str, token: str, username: str): """Create NextcloudClient with OAuth bearer token. Args: base_url: Nextcloud base URL token: OAuth access token username: Nextcloud username Returns: NextcloudClient configured with bearer token authentication """ from ..auth import BearerAuth # noqa: PLC0415 logger.info("Creating NC Client for user '%s' using OAuth token", username) return cls( base_url=base_url, username=username, auth=BearerAuth(token), token=token, ) async def capabilities(self): response = await self._client.get( "/ocs/v2.php/cloud/capabilities", headers={"OCS-APIRequest": "true", "Accept": "application/json"}, ) response.raise_for_status() return response.json() async def get_enabled_apps(self) -> set[str]: """Return the set of app ids enabled for the authenticated user. Uses the per-user core navigation endpoint, which lists only apps the current user can access (respecting group restrictions). The vector scanner uses this to skip polling apps the user lacks, which would 404 and flood tenant logs. Preferred over ``/cloud/capabilities`` because the News app advertises no capability and so never appears there. """ response = await self._client.get( "/ocs/v2.php/core/navigation/apps", headers={"OCS-APIRequest": "true", "Accept": "application/json"}, ) response.raise_for_status() data = response.json() # ``X or {}``/``or []`` (not ``.get(k, default)``) so a present-but-null # ``ocs``/``data`` (``{"ocs": null}``) coerces to empty instead of # raising AttributeError on ``None.get``. ocs = data.get("ocs") or {} # A 200 carrying ``meta.status != "ok"`` is an OCS-level failure (auth / # permission edge cases) that ``raise_for_status`` can't see. Raise so # the scanner's ``_get_enabled_apps_or_none`` catches it and falls back # to scanning every app, rather than silently gating all apps off for a # cycle on an empty ``data``. Tolerate a missing/empty meta (our own # mocks, and any envelope that omits it). status = (ocs.get("meta") or {}).get("status") if status and status != "ok": # falsy (missing/None/"") tolerated raise ValueError(f"OCS navigation returned status={status!r}") entries = ocs.get("data") or [] enabled: set[str] = set() for entry in entries: # ``app`` is the canonical app id; ``id`` matches it for the apps we # gate. Union both so an unexpected nav-entry shape never hides an # enabled app — a false "disabled" would skip real indexing. for key in ("app", "id"): value = entry.get(key) if value: enabled.add(value) return enabled async def notes_search_notes(self, *, query: str): """Search notes using token-based matching with relevance ranking.""" all_notes = self.notes.get_all_notes() return await self._notes_search.search_notes(all_notes, query) async def find_files_by_tag( self, tag_name: str, mime_type_filter: str | None = None ) -> list[dict]: """Return files carrying ``tag_name``, expanding tagged folders into matching descendants when ``mime_type_filter`` is set.""" tag = await self.webdav.get_tag_by_name(tag_name) if not tag: logger.debug("Tag %r not found, returning empty list", tag_name) return [] items = await self.webdav.get_files_by_tag(tag["id"]) if not items: logger.debug("No items found with tag %r", tag_name) return [] logger.debug( "Found %d directly-tagged item(s) with tag %r", len(items), tag_name ) # Split into directly-tagged files vs tagged directories. by_id: dict[int, dict] = {} tagged_dirs: list[dict] = [] for item in items: if item.get("is_directory"): tagged_dirs.append(item) continue if mime_type_filter and not item.get("content_type", "").startswith( mime_type_filter ): continue file_id = item.get("id") if file_id is None: continue by_id[file_id] = item # Expand each tagged directory into its descendant files matching # the MIME filter. Skip when no MIME filter is set — see docstring. if mime_type_filter and tagged_dirs: for dir_info in tagged_dirs: dir_path = dir_info.get("path", "").strip("/") try: # find_all_by_type pages past Nextcloud's default ~100-result # SEARCH page so every tagged-folder descendant is discovered; # find_by_type would silently cap a large folder. descendants = await self.webdav.find_all_by_type( mime_type_filter, scope=dir_path ) except Exception as e: logger.warning( "Tag-based directory walk failed for %r (tag %r): %s; " "skipping descendants", dir_path, tag_name, e, ) continue added = 0 for d in descendants: if d.get("is_directory"): continue file_id = ( d.get("file_id") if d.get("file_id") is not None else d.get("id") ) if file_id is None: continue if file_id in by_id: # Directly-tagged entry already wins; keeps the # canonical shape from get_files_by_tag. continue by_id[file_id] = _normalise_search_result(d) added += 1 logger.debug( "Tag %r: directory %r expanded to %d descendant %s file(s)", tag_name, dir_path, added, mime_type_filter, ) files = list(by_id.values()) if mime_type_filter: logger.info( "Returning %d file(s) with tag %r (mime_type=%s, " "%d directly-tagged folder(s) expanded)", len(files), tag_name, mime_type_filter, len(tagged_dirs), ) else: logger.info("Returning %d file(s) with tag %r", len(files), tag_name) return files def _get_webdav_base_path(self) -> str: """Helper to get the base WebDAV path for the authenticated user.""" return f"/remote.php/dav/files/{self.username}" async def __aenter__(self): """Async context manager entry.""" return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager exit - closes all clients.""" await self.close() return False # Don't suppress exceptions async def close(self): """Close the HTTP client and CalDAV client.""" await self._client.aclose() await self.calendar.close()