feat(data): integrate vosfs data commands

CONTEXT.md

@@ -16,6 +16,14 @@ _Avoid_: Base URL, host, cluster
 Required, unique, user-facing handle for one **Science Platform Server**, used to reference it in configuration and commands.
 _Avoid_: Label, alias, key
 
+**VOSpace Service**:
+Remote astronomical storage service associated with a **Science Platform Server**.
+_Avoid_: Storage backend, filesystem
+
+**Storage Identifier**:
+Required, globally unique, user-facing handle for one **VOSpace Service**, used to address it in configuration, commands, and operands such as `arc:/home/user`. The reserved identifier `local` addresses the machine where the command runs.
+_Avoid_: Storage Name, Server Name, alias, key
+
 **Identity Provider (IDP)**:
 Organization that issues user identity for CANFAR authentication.
 Initial IDPs are `Canadian Astronomy Data Centre (CADC)` and `SKA Regional Centre Network (SRCNet)`.
@@ -69,6 +77,8 @@ _Avoid_: Resource profile, quota
 
 - A **CANFAR Science Platform** exposes one or more **Science Platform Servers**.
 - A **Science Platform Server** is identified by its **Server Name**; its IVOA URI is discovery metadata, and two Server Names may point at the same endpoint.
+- A **Science Platform Server** can expose multiple **VOSpace Services**.
+- A **VOSpace Service** is identified by its **Storage Identifier** and uses its parent **Science Platform Server**'s **Identity Provider (IDP)**.
 - An **Identity Provider (IDP)** can support one or more **Science Platform Servers**.
 - **Authentication** and **Platform** are separate seams with independent ownership.
 - An **Authentication Record** belongs to one **Identity Provider (IDP)**.

canfar/auth/oidc.py

@@ -7,7 +7,7 @@
 import time
 from collections.abc import Awaitable, Callable, Generator
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import httpx
 from authlib.integrations.base_client.errors import OAuthError
@@ -20,6 +20,8 @@
 if TYPE_CHECKING:
     from authlib.integrations.httpx_client import AsyncOAuth2Client
 
+    from canfar.models.config import Configuration
+
 log = get_logger(__name__)
 _BASIC_AUTH_METHOD = "client_secret_basic"
 
@@ -193,6 +195,67 @@ def _validated_refresh(refreshed: Any) -> dict[str, Any]:
     return dict(refreshed)
 
 
+def _refresh(
+    credential: OIDCCredential,
+) -> tuple[str, str, str, str] | None:
+    """Return complete literal refresh inputs when the record is eligible."""
+    if not credential.refreshable:
+        return None
+    return (
+        cast("str", credential.endpoints.token),
+        cast("str", credential.client.identity),
+        cast("SecretStr", credential.client.secret).get_secret_value(),
+        cast("SecretStr", credential.token.refresh).get_secret_value(),
+    )
+
+
+def _persist(
+    config: Configuration,
+    credential: OIDCCredential,
+    refreshed: dict[str, Any],
+) -> OIDCCredential:
+    """Validate and atomically persist refreshed OIDC state."""
+    previous_refresh = credential.token.refresh
+    previous_refresh_value = (
+        previous_refresh.get_secret_value() if previous_refresh is not None else None
+    )
+    returned_refresh = refreshed.get("refresh_token")
+    refresh_value = (
+        returned_refresh
+        if isinstance(returned_refresh, str) and returned_refresh
+        else previous_refresh_value
+    )
+    access_value = refreshed.get("access_token")
+    if not isinstance(access_value, str) or not access_value:
+        msg = "OIDC token refresh failed: malformed token response"
+        raise ValueError(msg)
+    try:
+        token = Token(
+            access=SecretStr(access_value),
+            refresh=SecretStr(refresh_value) if refresh_value is not None else None,
+            token_type=refreshed.get("token_type") or credential.token.token_type,
+            scope=refreshed.get("scope") or credential.token.scope,
+        )
+        expiry = Expiry(
+            access=refreshed.get("expires_at"),
+            refresh=(
+                None
+                if refresh_value != previous_refresh_value
+                else credential.expiry.refresh
+            ),
+        )
+    except (KeyError, TypeError, ValidationError):
+        msg = "OIDC token refresh failed: malformed token response"
+        raise ValueError(msg) from None
+
+    updated = credential.model_copy(update={"token": token, "expiry": expiry})
+    candidate = config.model_copy(deep=True)
+    candidate.update_credential(updated)
+    candidate.save()
+    config.update_credential(updated)
+    return updated
+
+
 async def refresh(
     url: str,
     identity: str,

canfar/auth/x509.py

@@ -25,6 +25,16 @@
 class CertificateError(ValueError):
     """Raised when an X.509 certificate cannot be used."""
 
+    def __init__(
+        self,
+        message: str,
+        *,
+        expired_at: datetime | None = None,
+    ) -> None:
+        """Initialize a certificate error with optional structured expiry."""
+        super().__init__(message)
+        self.expired_at = expired_at
+
 
 def _to_utc(value: datetime) -> datetime:
     """Return timezone aware datetime.
@@ -88,7 +98,7 @@ def assert_valid_dates(
             f"Certificate {destination} expired on {valid_until.isoformat()}; "
             f"current time {now_utc.isoformat()}."
         )
-        raise CertificateError(msg)
+        raise CertificateError(msg, expired_at=valid_until)
 
 
 def gather(

canfar/cli/data.py

@@ -0,0 +1,60 @@
+"""Mount the upstream data command application."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import typer
+from fsspec_cli import App
+from typer.core import TyperGroup
+from typer.main import get_group
+
+from canfar.storage import sources
+
+if TYPE_CHECKING:
+    from typer._click.core import Command, Context
+
+_DATA_GROUP_META_KEY = "canfar.data_group"
+
+
+def group() -> TyperGroup:
+    """Build the released upstream application with CANFAR policy.
+
+    Returns:
+        TyperGroup: The upstream command group bound to configured sources.
+    """
+    return get_group(
+        App(
+            sources(),
+            capabilities={"recursion": {"copy": True, "remove": False}},
+        ).typer_app
+    )
+
+
+class _DataGroup(TyperGroup):
+    """Resolve the embedded app lazily so imports perform no configuration I/O."""
+
+    @staticmethod
+    def _delegate(ctx: Context) -> TyperGroup:
+        resolved = ctx.meta.get(_DATA_GROUP_META_KEY)
+        if resolved is None:
+            resolved = group()
+            ctx.meta[_DATA_GROUP_META_KEY] = resolved
+        assert isinstance(resolved, TyperGroup)
+        return resolved
+
+    def list_commands(self, ctx: Context) -> list[str]:
+        """List the unchanged upstream commands."""
+        return self._delegate(ctx).list_commands(ctx)
+
+    def get_command(self, ctx: Context, cmd_name: str) -> Command | None:
+        """Resolve an unchanged upstream command."""
+        return self._delegate(ctx).get_command(ctx, cmd_name)
+
+
+data = typer.Typer(
+    cls=_DataGroup,
+    help="Operate on configured data sources.",
+    add_completion=False,
+    no_args_is_help=True,
+)

canfar/cli/login.py

@@ -66,7 +66,7 @@ def _login_flow(
 
     idp_info = get_idp(idp)
     try:
-        credential = authenticate_for_cli(idp_info, timeout=timeout)
+        credential = authenticate_for_cli(idp_info, timeout=timeout, force=force)
     except (ValueError, RuntimeError) as exc:
         get_console(stderr=True).print(f"[bold red]{exc}[/bold red]")
         raise typer.Exit(1) from exc

canfar/cli/login_auth.py

@@ -6,6 +6,7 @@
 import webbrowser
 from typing import TYPE_CHECKING, Any
 
+import humanize
 import segno
 from rich import progress as rich_progress
 
@@ -98,12 +99,14 @@ def authenticate_for_cli(
     idp_info: IdpInfo,
     *,
     timeout: int | None = None,
+    force: bool = False,
 ) -> AuthenticationCredential:
     """Acquire Authentication credentials interactively for CLI login.
 
     Args:
         idp_info: Built-in Identity Provider metadata.
         timeout: HTTP timeout in seconds for OIDC requests.
+        force: Obtain a new X.509 certificate instead of reusing an existing one.
 
     Returns:
         Saved-ready authentication credential without embedded server.
@@ -113,19 +116,38 @@ def authenticate_for_cli(
         RuntimeError: If OIDC discovery URL is missing for an OIDC IDP.
     """
     if idp_info.auth_mode == "x509":
-        return _authenticate_x509(idp_info.key)
+        return _authenticate_x509(idp_info.key, force=force)
     return _authenticate_oidc(idp_info, timeout=timeout)
 
 
-def _authenticate_x509(idp: str) -> X509Credential:
-    """Run interactive X509 certificate acquisition.
+def _authenticate_x509(idp: str, *, force: bool = False) -> X509Credential:
+    """Reuse a valid X.509 certificate or acquire a replacement.
 
     Args:
         idp: Canonical Identity Provider key.
+        force: Obtain a new certificate without inspecting the existing one.
 
     Returns:
         X509 credential record for persisted configuration.
     """
+    if not force:
+        try:
+            info = x509.inspect()
+        except x509.CertificateError as exc:
+            if exc.expired_at is not None:
+                age = humanize.naturaltime(exc.expired_at)
+                console_utils.get_console().print(
+                    f"[yellow]x509 certificate expired {age}[/yellow]"
+                )
+        except (FileNotFoundError, PermissionError, ValueError):
+            pass
+        else:
+            return X509Credential(
+                idp=idp,
+                path=info["path"],
+                expiry=float(info["expiry"] or 0.0),
+            )
+
     return x509.authenticate_credential(X509Credential(idp=idp, expiry=0.0))
 
 

canfar/cli/main.py

@@ -11,6 +11,7 @@
 from canfar.cli.auth import auth
 from canfar.cli.config import config
 from canfar.cli.create import create
+from canfar.cli.data import data
 from canfar.cli.delete import delete
 from canfar.cli.events import events
 from canfar.cli.image import image
@@ -115,7 +116,8 @@ def warning_writer(error: StructuredError) -> None:
     if ctx.invoked_subcommand is None:
         get_console().print(ctx.get_help())
         raise typer.Exit(0)
-    set_before_command(ctx, _emit_banner_for_command)
+    if ctx.invoked_subcommand != "data":
+        set_before_command(ctx, _emit_banner_for_command)
 
 
 cli: typer.Typer = typer.Typer(
@@ -161,6 +163,12 @@ def warning_writer(error: StructuredError) -> None:
     rich_help_panel="Auth Management",
 )
 
+cli.add_typer(
+    data,
+    name="data",
+    rich_help_panel="Data Management",
+)
+
 cli.add_typer(
     create,
     no_args_is_help=True,