Reject retired API credentials before use

Author: jdalton

socketdev/__init__.py

@@ -1,5 +1,7 @@
 import os
 from socketdev.core.api import API
+from socketdev.core.retired_token import reject_retired_token
+from socketdev.core.token_selection import select_token
 from socketdev.dependencies import Dependencies
 from socketdev.diffscans import DiffScans
 from socketdev.export import Export
@@ -53,21 +55,15 @@
 
 class socketdev:
     def __init__(self, token: Optional[str] = None, timeout: int = 1200, allow_unverified: bool = False, user_agent: Optional[str] = None):
-        # Try to get token from environment variables if not provided
-        if token is None:
-            token = (
-                os.getenv("SOCKET_SECURITY_API_TOKEN") or
-                os.getenv("SOCKET_SECURITY_API_KEY") or
-                os.getenv("SOCKET_API_KEY") or
-                os.getenv("SOCKET_API_TOKEN")
-            )
-        
+        token = select_token(token, os.environ)
         if token is None:
             raise ValueError(
                 "API token is required. Provide it as a parameter or set one of these environment variables: "
                 "SOCKET_SECURITY_API_TOKEN, SOCKET_SECURITY_API_KEY, SOCKET_API_KEY, SOCKET_API_TOKEN"
             )
 
+        reject_retired_token(token)
+
         self.api = API()
         self.token = token + ":"
         self.api.encode_key(self.token)

socketdev/core/api.py

@@ -1,5 +1,6 @@
 import base64
 from socketdev.log import log
+from socketdev.core.retired_token import reject_retired_basic_auth_token
 
 import requests
 from socketdev.core.classes import Response
@@ -29,6 +30,7 @@ def __init__(self):
         self.user_agent = None
 
     def encode_key(self, token: str):
+        reject_retired_basic_auth_token(token)
         self.encoded_key = base64.b64encode(token.encode()).decode("ascii")
 
     def set_timeout(self, timeout: int):

socketdev/core/retired_token.py

@@ -0,0 +1,33 @@
+import hashlib
+import hmac
+from typing import Optional
+
+RETIRED_TOKEN_SHA256 = "9cab58b24140ee12c08e265729d246f9200f8ef9b1d884f90c06c4e6b69adf18"
+RETIRED_TOKEN_ERROR = (
+    "This API token has been retired. Create a new API token in Socket "
+    "and update your configuration."
+)
+
+
+def _matches_retired_digest(token: Optional[str], digest: str) -> bool:
+    if token is None:
+        return False
+    token_digest = hashlib.sha256(token.strip().encode("utf-8")).hexdigest()
+    return hmac.compare_digest(token_digest, digest)
+
+
+def _reject_retired_token(token: Optional[str], digest: str) -> None:
+    if _matches_retired_digest(token, digest):
+        raise ValueError(RETIRED_TOKEN_ERROR)
+
+
+def _reject_basic_auth_token(token: str, digest: str) -> None:
+    _reject_retired_token(token.partition(":")[0], digest)
+
+
+def reject_retired_token(token: Optional[str]) -> None:
+    _reject_retired_token(token, RETIRED_TOKEN_SHA256)
+
+
+def reject_retired_basic_auth_token(token: str) -> None:
+    _reject_basic_auth_token(token, RETIRED_TOKEN_SHA256)

socketdev/core/token_selection.py

@@ -0,0 +1,19 @@
+from typing import Mapping, Optional
+
+TOKEN_ENVIRONMENT_VARIABLES = (
+    "SOCKET_SECURITY_API_TOKEN",
+    "SOCKET_SECURITY_API_KEY",
+    "SOCKET_API_KEY",
+    "SOCKET_API_TOKEN",
+)
+
+
+def select_token(
+    explicit_token: Optional[str], environment: Mapping[str, str]
+) -> Optional[str]:
+    if explicit_token is not None:
+        return explicit_token
+    return next(
+        (environment[name] for name in TOKEN_ENVIRONMENT_VARIABLES if environment.get(name)),
+        None,
+    )

tests/unit/test_retired_token.py

@@ -0,0 +1,85 @@
+import base64
+import hashlib
+
+import pytest
+
+from socketdev import socketdev
+from socketdev.core.api import API
+from socketdev.core.retired_token import (
+    RETIRED_TOKEN_ERROR,
+    _reject_basic_auth_token,
+    _reject_retired_token,
+)
+from socketdev.core.token_selection import TOKEN_ENVIRONMENT_VARIABLES, select_token
+
+
+def test_retired_token_rejection_redacts_token_and_digest():
+    token = "synthetic-retired-token"
+    digest = hashlib.sha256(token.encode()).hexdigest()
+
+    with pytest.raises(ValueError) as exc_info:
+        _reject_retired_token(token, digest)
+
+    assert str(exc_info.value) == RETIRED_TOKEN_ERROR
+    assert token not in str(exc_info.value)
+    assert digest not in str(exc_info.value)
+
+
+def test_sdk_constructor_accepts_safe_explicit_token_without_network_access():
+    sdk = socketdev(token="synthetic-safe-token")
+
+    assert sdk.token == "synthetic-safe-token:"
+
+
+def test_api_encode_key_preserves_each_input_byte_sequence():
+    bare_api = API()
+    suffixed_api = API()
+    bare_token = "synthetic-safe-token"
+    suffixed_token = "synthetic-safe-token:"
+
+    bare_api.encode_key(bare_token)
+    suffixed_api.encode_key(suffixed_token)
+
+    assert bare_api.encoded_key == base64.b64encode(bare_token.encode()).decode("ascii")
+    assert suffixed_api.encoded_key == base64.b64encode(suffixed_token.encode()).decode("ascii")
+
+
+def test_api_rejects_retired_basic_username_with_password_suffix():
+    token = "synthetic-retired-token"
+    digest = hashlib.sha256(token.encode()).hexdigest()
+
+    with pytest.raises(ValueError, match="token has been retired"):
+        _reject_basic_auth_token(f"{token}:password", digest)
+
+
+def test_token_selection_preserves_explicit_and_environment_precedence():
+    environment = {
+        name: f"token-{index}"
+        for index, name in enumerate(TOKEN_ENVIRONMENT_VARIABLES)
+    }
+
+    assert select_token("", environment) == ""
+    assert select_token("explicit", environment) == "explicit"
+    assert select_token(None, environment) == "token-0"
+    assert select_token(
+        None,
+        {TOKEN_ENVIRONMENT_VARIABLES[0]: "", TOKEN_ENVIRONMENT_VARIABLES[1]: "next"},
+    ) == "next"
+    assert select_token(None, {}) is None
+
+
+@pytest.mark.parametrize("token", [None, "different-token"])
+def test_nonretired_token_is_accepted(token):
+    digest = hashlib.sha256(b"synthetic-retired-token").hexdigest()
+    _reject_retired_token(token, digest)
+
+
+def test_retired_token_comparison_trims_whitespace():
+    digest = hashlib.sha256(b"synthetic-retired-token").hexdigest()
+    with pytest.raises(ValueError):
+        _reject_retired_token(" \tsynthetic-retired-token\n", digest)
+
+
+@pytest.mark.parametrize("name", TOKEN_ENVIRONMENT_VARIABLES)
+def test_each_environment_source_is_supported(name):
+    assert select_token(None, {name: "selected-token"}) == "selected-token"