Add digest auth option to downloader integration

Author: mossymaker

homeassistant/components/downloader/const.py

@@ -7,9 +7,12 @@
 DOMAIN = "downloader"
 DEFAULT_NAME = "Downloader"
 CONF_DOWNLOAD_DIR = "download_dir"
+ATTR_DIGEST_AUTH = "digest_auth"
 ATTR_FILENAME = "filename"
+ATTR_DIGEST_PASSWORD = "digest_password"
 ATTR_SUBDIR = "subdir"
 ATTR_URL = "url"
+ATTR_DIGEST_USERNAME = "digest_username"
 ATTR_OVERWRITE = "overwrite"
 
 CONF_DOWNLOAD_DIR = "download_dir"

homeassistant/components/downloader/services.py

@@ -8,6 +8,7 @@
 import threading
 
 import requests
+from requests.auth import HTTPDigestAuth
 import voluptuous as vol
 
 from homeassistant.core import HomeAssistant, ServiceCall, callback
@@ -17,6 +18,9 @@
 
 from .const import (
     _LOGGER,
+    ATTR_DIGEST_AUTH,
+    ATTR_DIGEST_PASSWORD,
+    ATTR_DIGEST_USERNAME,
     ATTR_FILENAME,
     ATTR_OVERWRITE,
     ATTR_SUBDIR,
@@ -40,6 +44,12 @@ def do_download() -> None:
         try:
             url = service.data[ATTR_URL]
 
+            digest_auth = service.data.get(ATTR_DIGEST_AUTH)
+
+            digest_username = service.data.get(ATTR_DIGEST_USERNAME)
+
+            digest_password = service.data.get(ATTR_DIGEST_PASSWORD)
+
             subdir = service.data.get(ATTR_SUBDIR)
 
             filename = service.data.get(ATTR_FILENAME)
@@ -52,7 +62,18 @@ def do_download() -> None:
 
             final_path = None
 
-            req = requests.get(url, stream=True, timeout=10)
+            auth = (
+                HTTPDigestAuth(digest_username, digest_password)
+                if digest_auth and digest_username and digest_password
+                else None
+            )
+
+            req = requests.get(
+                url=url,
+                stream=True,
+                timeout=10,
+                auth=auth,
+            )
 
             if req.status_code != HTTPStatus.OK:
                 _LOGGER.warning(
@@ -151,9 +172,12 @@ def async_setup_services(hass: HomeAssistant) -> None:
         download_file,
         schema=vol.Schema(
             {
+                vol.Optional(ATTR_DIGEST_AUTH, default=False): cv.boolean,
                 vol.Optional(ATTR_FILENAME): cv.string,
+                vol.Optional(ATTR_DIGEST_PASSWORD): cv.string,
                 vol.Optional(ATTR_SUBDIR): cv.string,
                 vol.Required(ATTR_URL): cv.url,
+                vol.Optional(ATTR_DIGEST_USERNAME): cv.string,
                 vol.Optional(ATTR_OVERWRITE, default=False): cv.boolean,
             }
         ),

homeassistant/components/downloader/services.yaml

@@ -5,6 +5,15 @@ download_file:
       example: "http://example.org/myfile"
       selector:
         text:
+    digest_auth:
+      selector:
+        boolean:
+    digest_username:
+      selector:
+        text:
+    digest_password:
+      selector:
+        text:
     subdir:
       example: "download_dir"
       selector:

homeassistant/components/downloader/strings.json

@@ -21,6 +21,18 @@
           "name": "[%key:common::config_flow::data::url%]",
           "description": "The URL of the file to download."
         },
+        "digest_auth": {
+          "name": "Use Digest Authentication",
+          "description": "Whether to use digest authentication."
+        },
+        "digest_username": {
+          "name": "Digest Authentication Username",
+          "description": "Username to use with digest authentication."
+        },
+        "digest_password": {
+          "name": "Digest Authentication Password",
+          "description": "Password to use with digest authentication."
+        },
         "subdir": {
           "name": "Subdirectory",
           "description": "Relative download path."

tests/components/downloader/test_action_digest_auth.py

@@ -0,0 +1,79 @@
+"""Test downloader action."""
+
+from http import HTTPStatus
+from unittest.mock import mock_open, patch
+
+import requests_mock
+
+from homeassistant.components.downloader.const import (
+    ATTR_DIGEST_AUTH,
+    ATTR_DIGEST_PASSWORD,
+    ATTR_DIGEST_USERNAME,
+    ATTR_URL,
+    CONF_DOWNLOAD_DIR,
+    DOMAIN,
+    SERVICE_DOWNLOAD_FILE,
+)
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+def digest_challenge_matcher(request):
+    """Match requests with no auth header."""
+    return "Authorization" not in request.headers
+
+
+def digest_response_matcher(request):
+    """Match requests with an auth header."""
+    return "Authorization" in request.headers
+
+
+async def test_digest_auth(hass: HomeAssistant) -> None:
+    """Test digest authentication in downloader action."""
+
+    TEST_URL = "http://example.com/protected-resource"
+
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_DOWNLOAD_DIR: "/test_dir",
+        },
+    )
+    config_entry.add_to_hass(hass)
+    with patch("os.path.isdir", return_value=True):
+        await hass.config_entries.async_setup(config_entry.entry_id)
+
+    with requests_mock.Mocker() as mock:
+        mock.get(
+            TEST_URL,
+            additional_matcher=digest_challenge_matcher,
+            headers={"WWW-Authenticate": 'Digest realm="example.com", nonce="test"'},
+            status_code=HTTPStatus.UNAUTHORIZED,
+        )
+
+        mock.get(
+            TEST_URL,
+            additional_matcher=digest_response_matcher,
+            status_code=HTTPStatus.OK,
+        )
+
+        with patch("builtins.open", new=mock_open(), create=True):
+            await hass.services.async_call(
+                DOMAIN,
+                SERVICE_DOWNLOAD_FILE,
+                {
+                    ATTR_URL: TEST_URL,
+                    ATTR_DIGEST_AUTH: True,
+                    ATTR_DIGEST_USERNAME: "test",
+                    ATTR_DIGEST_PASSWORD: "test",
+                },
+                blocking=True,
+            )
+
+        assert mock.called
+
+        assert (
+            "Authorization" in mock.last_request.headers
+            and mock.last_request.headers["Authorization"].startswith("Digest")
+        )