Fix: Issue 195 path safety tests

cwmscli/commands/blob.py

@@ -11,10 +11,12 @@
 
 from cwmscli.utils import (
     colors,
+    format_local_download_error,
     get_api_key,
     has_invalid_chars,
     init_cwms_session,
     log_scoped_read_hint,
+    validate_default_download_dest,
 )
 from cwmscli.utils.click_help import DOCS_BASE_URL
 from cwmscli.utils.deps import requires
@@ -109,6 +111,14 @@ def _save_blob_content(
     return dest
 
 
+def _default_download_dest(blob_id: str) -> str:
+    return validate_default_download_dest(
+        blob_id,
+        resource_name="Blob",
+        docs_url=BLOB_DOCS_URL,
+    )
+
+
 def _blob_media_type(cwms_module, office: str, blob_id: str) -> Optional[str]:
     try:
         result = cwms_module.get_blobs(office_id=office, blob_id_like=blob_id)
@@ -602,7 +612,7 @@ def download_cmd(
 
     try:
         blob_content = cwms.get_blob(office_id=office, blob_id=bid)
-        target = dest or bid
+        target = dest or _default_download_dest(bid)
         _save_blob_content(
             blob_content,
             dest=target,
@@ -621,14 +631,15 @@ def download_cmd(
         )
         sys.exit(1)
     except Exception as e:
-        logging.error(f"Failed to download: {e}")
-        log_scoped_read_hint(
-            credential_kind=credential_kind,
-            anonymous=anonymous,
-            office=office,
-            action="download",
-            resource="blob content",
-        )
+        logging.error(format_local_download_error(e, BLOB_DOCS_URL))
+        if not isinstance(e, (OSError, ValueError)):
+            log_scoped_read_hint(
+                credential_kind=credential_kind,
+                anonymous=anonymous,
+                office=office,
+                action="download",
+                resource="blob content",
+            )
         sys.exit(1)
 
 

cwmscli/commands/clob.py

@@ -9,7 +9,13 @@
 import requests
 from cwms import api as cwms_api
 
-from cwmscli.utils import get_api_key, has_invalid_chars, log_scoped_read_hint
+from cwmscli.utils import (
+    format_local_download_error,
+    get_api_key,
+    has_invalid_chars,
+    log_scoped_read_hint,
+    validate_default_download_dest,
+)
 
 
 def _join_api_url(api_root: str, path: str) -> str:
@@ -29,6 +35,10 @@ def _write_clob_content(content: str, dest: str) -> str:
     return dest
 
 
+def _default_download_dest(clob_id: str) -> str:
+    return validate_default_download_dest(clob_id, resource_name="Clob")
+
+
 def _clob_endpoint_id(clob_id: str) -> tuple[str, Optional[str]]:
     normalized = clob_id.upper()
     if has_invalid_chars(normalized):
@@ -198,7 +208,7 @@ def download_cmd(
                 content = str(payload)
         else:
             content = _get_special_clob_text(office=office, clob_id=query_id)
-        target = dest or bid
+        target = dest or _default_download_dest(bid)
         _write_clob_content(content, target)
         logging.info(f"Downloaded clob to: {target}")
     except requests.HTTPError as e:
@@ -213,14 +223,7 @@ def download_cmd(
         )
         sys.exit(1)
     except Exception as e:
-        logging.error(f"Failed to download: {e}")
-        log_scoped_read_hint(
-            api_key=resolved_api_key,
-            anonymous=anonymous,
-            office=office,
-            action="download",
-            resource="clob content",
-        )
+        logging.error(format_local_download_error(e, ""))
         sys.exit(1)
 
 

cwmscli/utils/__init__.py

@@ -1,4 +1,5 @@
 import logging as py_logging
+import re
 import time
 from pathlib import Path
 from typing import Optional, Union
@@ -235,6 +236,66 @@ def log_scoped_read_hint(
     )
 
 
+def format_local_download_error(error: Exception, docs_url: str) -> str:
+    if isinstance(error, (OSError, ValueError)):
+        message = (
+            f"{colors.c('Failed to download:', 'red', bright=True)} {error}. "
+            f"If this is a local destination/path issue, pass "
+            f"{colors.c('--dest', 'cyan', bright=True)} explicitly."
+        )
+        if docs_url:
+            message = (
+                f"{message} {colors.c('Docs:', 'blue', bright=True)} "
+                f"{colors.c(docs_url, 'blue', bright=True)}"
+            )
+        return message
+    return f"{colors.c('Failed to download:', 'red', bright=True)} {error}"
+
+
+def validate_default_download_dest(
+    raw_id: str,
+    *,
+    resource_name: str,
+    docs_url: str = "",
+) -> str:
+    if raw_id is None:
+        raise ValueError(
+            f"{resource_name} ID must include a non-root destination name. "
+            f"Pass --dest explicitly if needed."
+        )
+
+    if raw_id.startswith("//") or raw_id.startswith("\\\\"):
+        raise ValueError(
+            f"{resource_name} ID must resolve to a relative local path. "
+            f"Pass --dest explicitly if needed."
+        )
+
+    target = raw_id.lstrip("/\\")
+    if not target:
+        message = (
+            f"{resource_name} ID must include a non-root destination name. "
+            f"Pass --dest explicitly if needed."
+        )
+        if docs_url:
+            message = f"{message} Docs: {docs_url}"
+        raise ValueError(message)
+
+    if re.match(r"^[A-Za-z]:", target):
+        raise ValueError(
+            f"{resource_name} ID must resolve to a relative local path. "
+            f"Pass --dest explicitly if needed."
+        )
+
+    parts = re.split(r"[\\/]", target)
+    if any(part in {"", ".", ".."} for part in parts):
+        raise ValueError(
+            f"{resource_name} ID must resolve to a relative local path. "
+            f"Pass --dest explicitly if needed."
+        )
+
+    return target
+
+
 def common_api_options(f):
     f = log_level_option(f)
     f = office_option(f)