Support new robot 7.4 secret variables as part of auth params

.github/workflows/pythonapp.yml

@@ -12,6 +12,9 @@ jobs:
       matrix:
         os: [ ubuntu-latest, macos-latest, windows-latest ]
         python-version: [ 3.8, 3.12 ]
+        # test with robot without and with Secret support? Not sure if
+        # it is worth it?
+        robot-version: [ 7.3.2, 7.4 ]
     steps:
       - uses: actions/checkout@v4
       - name: Set up Python
@@ -23,6 +26,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install -e .[test]
+          python -m pip install robotframework==${{ matrix.robot-version }}
       - name: Lint with flake8
         run: |
           # stop the build if there are Python syntax errors or undefined names
@@ -59,5 +63,5 @@ jobs:
         if: ${{ always() }}
         uses: actions/upload-artifact@v4
         with:
-          name: rf-tests-report-${{ matrix.os }}-${{ matrix.python-version }}
+          name: rf-tests-report-${{ matrix.os }}-${{ matrix.python-version }}-${{ matrix.robot-version }}
           path: ./tests-report

.gitignore

@@ -34,3 +34,4 @@ env/*
 
 # ignore http server log
 atests/http_server/http_server.log
+.claude/

atests/secretvar.py

@@ -0,0 +1,7 @@
+# inject secret into robot suite.. doing this via python
+# to ensure this can also run in older robot versions
+try:
+    from robot.api.types import Secret
+    SECRET_PASSWORD = Secret("secret_passwd")
+except (ImportError, ModuleNotFoundError):
+    SECRET_PASSWORD = "not-supported"

atests/test_authentication.robot

@@ -2,7 +2,7 @@
 Library     RequestsLibrary
 Library     customAuthenticator.py
 Resource    res_setup.robot
-
+Variables   secretvar.py
 
 *** Test Cases ***
 Get With Auth
@@ -23,12 +23,45 @@ Get With Custom Auth
 
 Get With Digest Auth
     [Tags]    get    get-cert
-    ${auth}=    Create List    user    pass
+    ${auth}=    Create List    user    passwd
     Create Digest Session
     ...    authsession
     ...    ${HTTP_LOCAL_SERVER}
     ...    auth=${auth}
     ...    debug=3
-    ${resp}=    GET On Session    authsession    /digest-auth/auth/user/pass
+    ${resp}=    GET On Session    authsession    /digest-auth/auth/user/passwd
+    Should Be Equal As Strings    ${resp.status_code}    200
+    Should Be Equal As Strings    ${resp.json()['authenticated']}    True
+
+Get With Auth with Robot Secrets
+    [Tags]     robot-74    get    get-cert
+    Skip If    $SECRET_PASSWORD == "not-supported"
+    ...    msg=robot version does not support secrets
+    ${auth}=    Create List    user     ${SECRET_PASSWORD}
+    Create Session    authsession    ${HTTP_LOCAL_SERVER}    auth=${auth}
+    ${resp}=    GET On Session    authsession    /basic-auth/user/secret_passwd
+    Should Be Equal As Strings    ${resp.status_code}    200
+    Should Be Equal As Strings    ${resp.json()['authenticated']}    True
+
+Get With Digest Auth with Robot Secrets
+    [Tags]    robot-74    get    get-cert
+    Skip If    $SECRET_PASSWORD == "not-supported"
+    ...    msg=robot version does not support secrets
+    ${auth}=    Create List    user    ${SECRET_PASSWORD}
+    Create Digest Session
+    ...    authsession
+    ...    ${HTTP_LOCAL_SERVER}
+    ...    auth=${auth}
+    ...    debug=3
+    ${resp}=    GET On Session    authsession    /digest-auth/auth/user/secret_passwd
+    Should Be Equal As Strings    ${resp.status_code}    200
+    Should Be Equal As Strings    ${resp.json()['authenticated']}    True
+
+Session-less GET With Auth with Robot Secrets
+    [Tags]    robot-74    get    get-cert    session-less
+    Skip If    $SECRET_PASSWORD == "not-supported"
+    ...    msg=robot version does not support secrets
+    ${auth}=    Create List    user    ${SECRET_PASSWORD}
+    ${resp}=    GET    ${HTTP_LOCAL_SERVER}/basic-auth/user/secret_passwd    auth=${auth}
     Should Be Equal As Strings    ${resp.status_code}    200
     Should Be Equal As Strings    ${resp.json()['authenticated']}    True

src/RequestsLibrary/RequestsKeywords.py

@@ -6,6 +6,7 @@
 from RequestsLibrary import log
 from RequestsLibrary.compat import urljoin
 from RequestsLibrary.utils import (
+    check_and_process_secrets,
     is_list_or_tuple,
     is_file_descriptor,
     warn_if_equal_symbol_in_url_session_less,
@@ -22,6 +23,14 @@ def __init__(self):
         self.timeout = None
         self.cookies = None
         self.last_response = None
+        self._request_has_secrets = False
+        self._session_secrets = {}  # Maps session object ID to secrets flag
+
+    def _get_session_secrets_flag(self, session):
+        """Get the secrets flag for a session object"""
+        if not session:
+            return False
+        return self._session_secrets.get(id(session), False)
 
     def _common_request(self, method, session, uri, **kwargs):
 
@@ -30,6 +39,19 @@ def _common_request(self, method, session, uri, **kwargs):
         else:
             request_function = getattr(requests, "request")
 
+        auth = kwargs.get("auth")
+        if auth is not None and isinstance(auth, (list, tuple)):
+            kwargs["auth"], contains_secrets = check_and_process_secrets(auth)
+        else:
+            contains_secrets = False
+
+        if session:
+            # Check if the session was created with robot secrets
+            contains_secrets = contains_secrets or self._get_session_secrets_flag(session)
+
+        # Store secrets flag for _print_debug to access
+        self._request_has_secrets = contains_secrets
+
         self._capture_output()
 
         resp = request_function(
@@ -40,7 +62,7 @@ def _common_request(self, method, session, uri, **kwargs):
             **kwargs
         )
 
-        log.log_request(resp)
+        log.log_request(resp, has_secrets=contains_secrets)
         self._print_debug()
 
         log.log_response(resp)
@@ -59,7 +81,7 @@ def _close_file_descriptors(files, data):
         """
         Helper method that closes any open file descriptors.
         """
-        
+
         if is_list_or_tuple(files):
             files_descriptor_to_close = filter(
                 is_file_descriptor, [file[1][1] for file in files] + [data]
@@ -68,10 +90,10 @@ def _close_file_descriptors(files, data):
             files_descriptor_to_close = filter(
                 is_file_descriptor, list(files.values()) + [data]
             )
-        
+
         for file_descriptor in files_descriptor_to_close:
             file_descriptor.close()
-    
+
     @staticmethod
     def _merge_url(session, uri):
         """

src/RequestsLibrary/SessionKeywords.py

@@ -1,4 +1,5 @@
 import logging
+import re
 import sys
 
 import requests
@@ -12,7 +13,8 @@
 from RequestsLibrary import utils
 from RequestsLibrary.compat import RetryAdapter, httplib
 from RequestsLibrary.exceptions import InvalidExpectedStatus, InvalidResponse
-from RequestsLibrary.utils import is_string_type
+from RequestsLibrary.log import AUTHORIZATION
+from RequestsLibrary.utils import is_string_type, check_and_process_secrets
 
 from .RequestsKeywords import RequestsKeywords
 
@@ -25,6 +27,10 @@
 class SessionKeywords(RequestsKeywords):
     DEFAULT_RETRY_METHOD_LIST = RetryAdapter.get_default_allowed_methods()
 
+    def _set_session_secrets_flag(self, session, has_secrets):
+        """Store the secrets flag for a session object using its id as key"""
+        self._session_secrets[id(session)] = has_secrets
+
     def _create_session(
         self,
         alias,
@@ -172,15 +178,19 @@ def create_session(
                               Note that max_retries must be greater than 0.
 
         """
-        auth = requests.auth.HTTPBasicAuth(*auth) if auth else None
+        if auth:
+            processed_auth, session_has_secrets = check_and_process_secrets(auth)
+            auth = requests.auth.HTTPBasicAuth(*processed_auth)
+        else:
+            session_has_secrets = False
 
         logger.info(
             "Creating Session using : alias=%s, url=%s, headers=%s, \
                     cookies=%s, auth=%s, timeout=%s, proxies=%s, verify=%s, \
                     debug=%s "
             % (alias, url, headers, cookies, auth, timeout, proxies, verify, debug)
         )
-        return self._create_session(
+        session = self._create_session(
             alias=alias,
             url=url,
             headers=headers,
@@ -196,6 +206,8 @@ def create_session(
             retry_status_list=retry_status_list,
             retry_method_list=retry_method_list,
         )
+        self._set_session_secrets_flag(session, session_has_secrets)
+        return session
 
     @keyword("Create Client Cert Session")
     def create_client_cert_session(
@@ -262,7 +274,11 @@ def create_client_cert_session(
                               eg. set to [502, 503] to retry requests if those status are returned.
                               Note that max_retries must be greater than 0.
         """
-        auth = requests.auth.HTTPBasicAuth(*auth) if auth else None
+        if auth:
+            processed_auth, session_has_secrets = check_and_process_secrets(auth)
+            auth = requests.auth.HTTPBasicAuth(*processed_auth)
+        else:
+            session_has_secrets = False
 
         logger.info(
             "Creating Session using : alias=%s, url=%s, headers=%s, \
@@ -300,6 +316,7 @@ def create_client_cert_session(
         )
 
         session.cert = tuple(client_certs)
+        self._set_session_secrets_flag(session, session_has_secrets)
         return session
 
     @keyword("Create Custom Session")
@@ -452,9 +469,14 @@ def create_digest_session(
                               eg. set to [502, 503] to retry requests if those status are returned.
                               Note that max_retries must be greater than 0.
         """
-        digest_auth = requests.auth.HTTPDigestAuth(*auth) if auth else None
+        if auth:
+            processed_auth, session_has_secrets = check_and_process_secrets(auth)
+            digest_auth = requests.auth.HTTPDigestAuth(*processed_auth)
+        else:
+            digest_auth = None
+            session_has_secrets = False
 
-        return self._create_session(
+        session = self._create_session(
             alias=alias,
             url=url,
             headers=headers,
@@ -470,6 +492,8 @@ def create_digest_session(
             retry_status_list=retry_status_list,
             retry_method_list=retry_method_list,
         )
+        self._set_session_secrets_flag(session, session_has_secrets)
+        return session
 
     @keyword("Create Ntlm Session")
     def create_ntlm_session(
@@ -543,7 +567,8 @@ def create_ntlm_session(
                 " - expected 3, got {}".format(len(auth))
             )
         else:
-            ntlm_auth = HttpNtlmAuth("{}\\{}".format(auth[0], auth[1]), auth[2])
+            processed_auth, session_has_secrets = check_and_process_secrets(auth)
+            ntlm_auth = HttpNtlmAuth("{}\\{}".format(processed_auth[0], processed_auth[1]), processed_auth[2])
             logger.info(
                 "Creating NTLM Session using : alias=%s, url=%s, \
                         headers=%s, cookies=%s, ntlm_auth=%s, timeout=%s, \
@@ -561,7 +586,7 @@ def create_ntlm_session(
                 )
             )
 
-            return self._create_session(
+            session = self._create_session(
                 alias=alias,
                 url=url,
                 headers=headers,
@@ -577,6 +602,8 @@ def create_ntlm_session(
                 retry_status_list=retry_status_list,
                 retry_method_list=retry_method_list,
             )
+            self._set_session_secrets_flag(session, session_has_secrets)
+            return session
 
     @keyword("Session Exists")
     def session_exists(self, alias):
@@ -656,4 +683,14 @@ def _print_debug(self):
             debug_info = "\n".join(
                 [ll.rstrip() for ll in debug_info.splitlines() if ll.strip()]
             )
+
+            # Mask Authorization header in debug output when secrets are used
+            if self._request_has_secrets:
+                debug_info = re.sub(
+                    rf'({AUTHORIZATION}:)\s*([^\n]+)',
+                    r'\1 *****',
+                    debug_info,
+                    flags=re.IGNORECASE
+                )
+
             logger.debug(debug_info)

src/RequestsLibrary/log.py

@@ -17,17 +17,22 @@ def log_response(response):
     )
 
 
-def log_request(response):
+def log_request(response, has_secrets=False):
     request = response.request
     if response.history:
         original_request = response.history[0].request
         redirected = "(redirected) "
     else:
         original_request = request
         redirected = ""
+
+    # Mask Authorization header based on whether secrets were used
     safe_headers = dict(original_request.headers)
-    if logger.LOGLEVEL not in ['TRACE', 'DEBUG'] and AUTHORIZATION in safe_headers:
-        safe_headers[AUTHORIZATION] = '*****'
+    if AUTHORIZATION in safe_headers:
+        # If secrets were used, always mask. Otherwise, only mask if not in DEBUG/TRACE
+        if has_secrets or logger.LOGLEVEL not in ['TRACE', 'DEBUG']:
+            safe_headers[AUTHORIZATION] = '*****'
+
     logger.info(
         "%s Request : " % original_request.method.upper()
         + "url=%s %s\n " % (original_request.url, redirected)

src/RequestsLibrary/utils.py

@@ -5,6 +5,11 @@
 from requests.status_codes import codes
 from requests.structures import CaseInsensitiveDict
 from robot.api import logger
+try:
+    from robot.api.types import Secret
+    robot_supports_secrets = True
+except (ImportError, ModuleNotFoundError):
+    robot_supports_secrets = False
 
 from RequestsLibrary.compat import urlencode
 from RequestsLibrary.exceptions import UnknownStatusError
@@ -74,9 +79,35 @@ def is_string_type(data):
 def is_file_descriptor(fd):
     return isinstance(fd, io.IOBase)
 
+
 def is_list_or_tuple(data):
     return isinstance(data, (list, tuple))
 
+
+def check_and_process_secrets(auth):
+    """
+    Check if auth contains secrets and process them
+
+    Returns:
+        tuple: (processed_auth, has_secrets_flag)
+    """
+    if not auth or not isinstance(auth, (list, tuple)):
+        return auth, False
+
+    if robot_supports_secrets:
+        has_secrets_flag = False
+        processed = []
+        for a in auth:
+            if isinstance(a, Secret):
+                has_secrets_flag = True
+                processed.append(a.value)
+            else:
+                processed.append(a)
+        return tuple(processed), has_secrets_flag
+    else:
+        return auth, False
+
+
 def utf8_urlencode(data):
     if is_string_type(data):
         return data.encode("utf-8")