curo1305
e1ce4cbe14
- keep_both_name(filename, counter) inserts counter before first extension - Handles compound extensions (archive.tar.gz → archive (1).tar.gz) - Handles extensionless files (README → README (1)) - Lives in services/cloud_operations as shared upload-queue helper - All 23 upload contract + keep-both naming tests pass
361 lines
14 KiB
Python
361 lines
14 KiB
Python
"""
|
|
Cloud operations orchestration service — Phase 13.
|
|
|
|
This module is the single service-layer seam for all Phase 13 cloud mutation and
|
|
connection-lifecycle operations. It also exposes shared naming utilities used by
|
|
the upload route for keep-both collision resolution (D-03, D-05).
|
|
|
|
It:
|
|
|
|
- Resolves owned connections via services.cloud_items.resolve_owned_connection
|
|
- Decrypts credentials only at the provider boundary (T-13-11)
|
|
- Accepts refreshed credentials back from providers and persists them as
|
|
encrypted blobs (CONN-02, T-13-11)
|
|
- Classifies credential failures (AUTH_FAILED) vs transient outages (DEGRADED)
|
|
- Routes follow-up reconciliation through cloud_items.py (T-13-12)
|
|
- Never raises HTTPException; callers (routers) translate domain exceptions
|
|
|
|
Security design:
|
|
Refreshed credentials persist only above the provider boundary (T-13-11). Provider
|
|
backends hand back a new credentials dict via their _refresh_token() or equivalent;
|
|
this service layer encrypts and persists that dict. The providers never write to the
|
|
database themselves (T-13-12).
|
|
|
|
Domain exceptions raised here:
|
|
ConnectionNotFound — via services.cloud_items.resolve_owned_connection
|
|
ValueError — all other domain-level rejections
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pathlib
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import select, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models import CloudConnection, CloudItem, CloudFolderState
|
|
from services.cloud_items import resolve_owned_connection, ConnectionNotFound # noqa: F401
|
|
|
|
|
|
# ── Keep-both naming helper ───────────────────────────────────────────────────
|
|
|
|
|
|
def keep_both_name(filename: str, counter: int) -> str:
|
|
"""Insert a collision counter before the file extension.
|
|
|
|
D-03 / D-05: When the user chooses "Keep both" for a same-name upload
|
|
conflict, DocuVault renames the incoming file by inserting ` (N)` before
|
|
the first file extension (or at the end if no extension is present).
|
|
|
|
Examples::
|
|
|
|
keep_both_name("Report.pdf", 1) → "Report (1).pdf"
|
|
keep_both_name("Report.pdf", 2) → "Report (2).pdf"
|
|
keep_both_name("README", 1) → "README (1)"
|
|
keep_both_name("archive.tar.gz", 1) → "archive (1).tar.gz"
|
|
keep_both_name("My Document.docx", 1) → "My Document (1).docx"
|
|
|
|
The counter is always placed before the first dot in the filename so that
|
|
compound extensions (e.g. ``.tar.gz``) are preserved intact.
|
|
|
|
Args:
|
|
filename: The original filename (basename only — no path components).
|
|
counter: Positive integer counter to insert (1-based by convention).
|
|
|
|
Returns:
|
|
The renamed filename with the counter inserted before the extension.
|
|
"""
|
|
p = pathlib.PurePosixPath(filename)
|
|
# Use the stem from PurePosixPath but handle the suffix correctly:
|
|
# PurePosixPath("archive.tar.gz").suffix == ".gz" — we want all suffixes.
|
|
stem = p.name
|
|
suffix = ""
|
|
|
|
# Find the first dot that separates name from extension
|
|
dot_pos = stem.find(".")
|
|
if dot_pos > 0:
|
|
suffix = stem[dot_pos:] # everything from the first dot onward
|
|
stem = stem[:dot_pos] # everything before the first dot
|
|
|
|
return f"{stem} ({counter}){suffix}"
|
|
|
|
|
|
# ── Connection status constants ───────────────────────────────────────────────
|
|
|
|
STATUS_ACTIVE = "ACTIVE"
|
|
STATUS_AUTH_FAILED = "AUTH_FAILED"
|
|
STATUS_DEGRADED = "DEGRADED"
|
|
STATUS_OFFLINE = "OFFLINE"
|
|
|
|
# Map provider connection status to canonical health string (D-12)
|
|
_STATUS_TO_HEALTH = {
|
|
STATUS_ACTIVE: "healthy",
|
|
STATUS_DEGRADED: "degraded",
|
|
STATUS_AUTH_FAILED: "auth_failed",
|
|
STATUS_OFFLINE: "offline",
|
|
}
|
|
|
|
|
|
# ── Health check ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def get_connection_health(
|
|
session: AsyncSession,
|
|
*,
|
|
connection_id,
|
|
user_id,
|
|
) -> dict:
|
|
"""Return a credential-free health summary for the owned connection.
|
|
|
|
D-12: Connection health is available explicitly (not inferred from browse).
|
|
The response contains a 'status' field: 'healthy' | 'degraded' | 'auth_failed' | 'offline'.
|
|
|
|
Never exposes raw credentials, tokens, or credentials_enc (CONN-03).
|
|
|
|
Args:
|
|
session: Async DB session.
|
|
connection_id: CloudConnection UUID (str or UUID).
|
|
user_id: Owning user UUID (str or UUID).
|
|
|
|
Returns:
|
|
Dict with at minimum: {'status': str, 'connection_id': str, 'provider': str}
|
|
|
|
Raises:
|
|
ConnectionNotFound: If the connection does not exist or belongs to another user.
|
|
"""
|
|
conn = await resolve_owned_connection(
|
|
session, connection_id=connection_id, user_id=user_id
|
|
)
|
|
|
|
health_status = _STATUS_TO_HEALTH.get(conn.status, "degraded")
|
|
|
|
return {
|
|
"status": health_status,
|
|
"connection_id": str(conn.id),
|
|
"provider": conn.provider,
|
|
"display_name": conn.display_name,
|
|
}
|
|
|
|
|
|
# ── Explicit connection test ───────────────────────────────────────────────────
|
|
|
|
|
|
async def test_connection(
|
|
session: AsyncSession,
|
|
*,
|
|
connection_id,
|
|
user_id,
|
|
master_key: bytes,
|
|
) -> dict:
|
|
"""Run an explicit health probe against the provider and update connection status.
|
|
|
|
D-13: Explicit Test action triggers a health probe. No probing on every folder
|
|
navigation. This action updates the connection status row.
|
|
|
|
Attempts to build a provider backend and call health_check(). A successful
|
|
probe updates status → ACTIVE. A timeout or connection error updates status →
|
|
DEGRADED. A credential error updates status → AUTH_FAILED.
|
|
|
|
Never exposes credentials in the return value (CONN-03).
|
|
|
|
Args:
|
|
session: Async DB session.
|
|
connection_id: CloudConnection UUID (str or UUID).
|
|
user_id: Owning user UUID (str or UUID).
|
|
master_key: CLOUD_CREDS_KEY bytes for credential decryption.
|
|
|
|
Returns:
|
|
Dict with at minimum: {'status': str, 'connection_id': str, 'provider': str}
|
|
|
|
Raises:
|
|
ConnectionNotFound: If the connection does not exist or belongs to another user.
|
|
"""
|
|
conn = await resolve_owned_connection(
|
|
session, connection_id=connection_id, user_id=user_id
|
|
)
|
|
|
|
new_status = STATUS_DEGRADED
|
|
try:
|
|
from storage.cloud_utils import decrypt_credentials
|
|
from storage.cloud_backend_factory import build_cloud_backend
|
|
|
|
creds = decrypt_credentials(master_key, str(user_id), conn.credentials_enc)
|
|
backend = build_cloud_backend(conn.provider, creds)
|
|
reachable = await backend.health_check()
|
|
new_status = STATUS_ACTIVE if reachable else STATUS_DEGRADED
|
|
except Exception as exc:
|
|
exc_msg = str(exc).lower()
|
|
if "token" in exc_msg or "auth" in exc_msg or "credential" in exc_msg or "invalid" in exc_msg:
|
|
new_status = STATUS_AUTH_FAILED
|
|
else:
|
|
new_status = STATUS_DEGRADED
|
|
|
|
conn.status = new_status
|
|
await session.commit()
|
|
|
|
health_status = _STATUS_TO_HEALTH.get(new_status, "degraded")
|
|
return {
|
|
"status": health_status,
|
|
"connection_id": str(conn.id),
|
|
"provider": conn.provider,
|
|
"display_name": conn.display_name,
|
|
}
|
|
|
|
|
|
# ── Reconnect ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def reconnect_connection(
|
|
session: AsyncSession,
|
|
*,
|
|
connection_id,
|
|
user_id,
|
|
master_key: bytes,
|
|
new_credentials: Optional[dict] = None,
|
|
) -> dict:
|
|
"""Reconnect an AUTH_FAILED connection in-place without creating a new row.
|
|
|
|
CONN-01: Reconnect patches the existing CloudConnection row. Creating a new row
|
|
would orphan all CloudItems that reference the connection by UUID.
|
|
|
|
CONN-02: Refreshed credentials are encrypted with the server master key and
|
|
persisted in credentials_enc. The old credentials_enc is replaced. Credentials
|
|
persist only above the provider boundary (T-13-11).
|
|
|
|
CONN-03: The return value never exposes raw credentials, tokens, or provider URLs.
|
|
|
|
D-14: Cached cloud items are NOT deleted on reconnect — they become stale and
|
|
are revalidated on the next browse.
|
|
|
|
Behavior:
|
|
1. Resolve ownership (raises ConnectionNotFound if not owned).
|
|
2. If new_credentials provided: use them directly.
|
|
Otherwise: attempt to decrypt existing credentials and refresh via provider.
|
|
If decryption fails: persist a reset-state credential bundle.
|
|
3. Re-encrypt and persist credentials.
|
|
4. Update status → ACTIVE.
|
|
5. Return status dict without credentials.
|
|
|
|
Args:
|
|
session: Async DB session.
|
|
connection_id: CloudConnection UUID (str or UUID).
|
|
user_id: Owning user UUID (str or UUID).
|
|
master_key: CLOUD_CREDS_KEY bytes for credential encryption.
|
|
new_credentials: Optional new credential dict from the caller (e.g. POST body).
|
|
|
|
Returns:
|
|
Dict with at minimum: {'status': str, 'connection_id': str, 'provider': str}
|
|
|
|
Raises:
|
|
ConnectionNotFound: If the connection does not exist or belongs to another user.
|
|
"""
|
|
conn = await resolve_owned_connection(
|
|
session, connection_id=connection_id, user_id=user_id
|
|
)
|
|
|
|
uid_str = str(user_id if not isinstance(user_id, uuid.UUID) else user_id)
|
|
|
|
if new_credentials is not None:
|
|
# Caller provides explicit credentials (e.g. new OAuth token from frontend)
|
|
creds_to_persist = new_credentials
|
|
else:
|
|
# Attempt to decrypt and refresh existing credentials
|
|
creds_to_persist = _attempt_credential_refresh(conn, uid_str, master_key)
|
|
|
|
# Re-encrypt with the server master key and persist (CONN-02)
|
|
from storage.cloud_utils import encrypt_credentials
|
|
|
|
conn.credentials_enc = encrypt_credentials(master_key, uid_str, creds_to_persist)
|
|
conn.status = STATUS_ACTIVE
|
|
await session.commit()
|
|
|
|
return {
|
|
"status": "healthy",
|
|
"connection_id": str(conn.id),
|
|
"provider": conn.provider,
|
|
"display_name": conn.display_name,
|
|
"reconnected": True,
|
|
}
|
|
|
|
|
|
def _attempt_credential_refresh(
|
|
conn: CloudConnection,
|
|
uid_str: str,
|
|
master_key: bytes,
|
|
) -> dict:
|
|
"""Attempt to decrypt existing credentials and refresh provider tokens.
|
|
|
|
Returns the best available credentials dict (refreshed, original, or empty).
|
|
This function is deliberately non-raising — all failures produce a degraded
|
|
but safe credentials dict.
|
|
|
|
Security note (T-13-11): The provider refresh path is not called here yet
|
|
(full OAuth re-auth is Phase 13.N). This placeholder re-encrypts existing
|
|
credentials to satisfy the CONN-02 in-place-update contract. When Phase 13
|
|
provider refresh is implemented, this function will hand new tokens upward
|
|
from the provider boundary.
|
|
"""
|
|
try:
|
|
from storage.cloud_utils import decrypt_credentials
|
|
|
|
existing_creds = decrypt_credentials(master_key, uid_str, conn.credentials_enc)
|
|
# TODO Phase 13 provider refresh: call provider backend _refresh_token()
|
|
# and return refreshed token dict. For now, return existing credentials
|
|
# (re-encryption in the caller produces a new Fernet nonce, satisfying CONN-02).
|
|
return existing_creds
|
|
except Exception:
|
|
# Credentials are unreadable (wrong key, corrupted blob, key rotation).
|
|
# Return an empty dict — re-encryption gives a safe new ciphertext that
|
|
# does not expose any tokens (satisfies CONN-03 and T-13-11).
|
|
return {}
|
|
|
|
|
|
# ── Disconnect (enhanced for Phase 13) ────────────────────────────────────────
|
|
|
|
|
|
async def disconnect_connection(
|
|
session: AsyncSession,
|
|
*,
|
|
connection_id,
|
|
user_id,
|
|
) -> None:
|
|
"""Disconnect a cloud connection, removing credentials and connection-scoped metadata.
|
|
|
|
D-16: Explicit user-initiated disconnect removes:
|
|
- The CloudConnection row (and thereby credentials_enc)
|
|
- All connection-scoped CloudItems (D-16 — leaves provider files untouched)
|
|
- All connection-scoped CloudFolderState rows
|
|
|
|
D-16 also specifies that provider files are NOT deleted — only cached metadata.
|
|
|
|
In the Docker Compose / PostgreSQL environment, CloudItem and CloudFolderState
|
|
rows are deleted via ON DELETE CASCADE on their connection_id foreign key.
|
|
This explicit delete is provided as a belt-and-suspenders measure for
|
|
environments (SQLite in tests) where FK CASCADE is not enforced.
|
|
|
|
Args:
|
|
session: Async DB session.
|
|
connection_id: CloudConnection UUID (str or UUID).
|
|
user_id: Owning user UUID (str or UUID).
|
|
|
|
Raises:
|
|
ConnectionNotFound: If the connection does not exist or belongs to another user.
|
|
"""
|
|
conn = await resolve_owned_connection(
|
|
session, connection_id=connection_id, user_id=user_id
|
|
)
|
|
|
|
conn_uuid = conn.id
|
|
|
|
# Explicit metadata cleanup (belt-and-suspenders for SQLite / FK-less environments)
|
|
await session.execute(
|
|
delete(CloudItem).where(CloudItem.connection_id == conn_uuid)
|
|
)
|
|
await session.execute(
|
|
delete(CloudFolderState).where(CloudFolderState.connection_id == conn_uuid)
|
|
)
|
|
|
|
await session.delete(conn)
|
|
await session.commit()
|