Merge pull request #2634 from sebix/fix-http-auth

http mixin: only send auth when strings not empty

Author: sebix

CHANGELOG.md

@@ -22,6 +22,7 @@ Please refer to the [NEWS](NEWS.md) for a list of changes which have an affect o
 - `intelmq.lib.mixins.sql` Add Support for MySQL (PR#2625 by Karl-Johan Karlsson).
 - New parameter `stop_retry_limit` to gracefully handle stopping bots which take longer to shutdown (PR#2598 by Lukas Heindl, fixes #2595).
 - `intelmq.lib.datatypes`: Remove unneeded Dict39 alias (PR#2639 by Nakul Rajpal, fixes #2635)
+- `intelmq.lib.mixins.http`: Only set HTTP header 'Authorization' if username or password are set and are not both empty string as they are by default in the Manager (fixes #2590, PR#2634 by Sebastian Wagner).
 
 ### Development
 

docs/user/bots.md

@@ -4520,11 +4520,11 @@ Using 'intelmq' as the `elastic_index`, the following are examples of the genera
 
 **`http_username`**
 
-(optional, string) HTTP basic authentication username.
+(optional, string) HTTP basic authentication username. Also set `http_password`.
 
 **`http_password`**
 
-(optional, string) HTTP basic authentication password.
+(optional, string) HTTP basic authentication password. Also set `http_username`.
 
 **`use_ssl`**
 

intelmq/lib/mixins/http.py

@@ -65,9 +65,12 @@ def setup(self):
         # tls certificate settings
         if self.ssl_client_cert is not None:
             self.__session.cert = self.ssl_client_cert
-        # auth settings
-        if self.http_username is not None:
+        # auth settings. username or password must exist (if only one, then this is likely an error, but we should try without auth) and not be an empty string
+        if self.http_username and self.http_password:
             self.__auth = (self.http_username, self.http_password)
+        elif self.http_username or self.http_password:
+            # only one, but not both are given
+            self.logger.warning("Either 'http_username' or 'http_password' are given, but for HTTP Authentication, both must be set.")
         self.__session.auth = self.__auth
         # headers settings
         if self.http_header is not None:

intelmq/tests/bots/collectors/http/test_collector.py

@@ -196,5 +196,66 @@ def test_debug_request_response_log(self, mocker):
         self.assertLogMatches("Response body: 'Should be in logs'.", 'DEBUG')
 
 
+@requests_mock.Mocker()
+class TestHTTPCollectorBotAuthentication(test.BotTestCase, unittest.TestCase):
+    """
+    HttpMixin setup is called a initialization
+    """
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = HTTPCollectorBot
+        cls.sysconfig = {'http_url': 'http://localhost/foobar',
+                         'name': 'Example feed',
+                         'http_username': '',
+                         'http_password': '',
+                         'logging_level': 'DEBUG'
+                         }
+
+    def test_empty_auth(self, mocker):
+        """
+        Test that no Authorization is set when username and password are empty strings.
+        """
+        def check_authorization_header(request):
+            print(f'check_authorization_header: {request.headers}')  # show the erroneos headers when the test fails
+            return 'Authorization' not in request.headers
+        # generates a mock address if the Authorization header is empty, otherwise the test fails
+        captured = mocker.register_uri('GET', self.sysconfig['http_url'],
+                            text='Foo Bar',
+                            additional_matcher=check_authorization_header)
+        self.run_bot()
+
+    def test_auth_only_username_or_password(self, mocker):
+        """
+        Test that no Authorization is set only the username is given and password is missing
+        """
+        def check_authorization_header(request):
+            print(f'check_authorization_header: {request.headers}')  # show the erroneos headers when the test fails
+            return 'Authorization' not in request.headers
+        # generates a mock address if the Authorization header is empty, otherwise the test fails
+        captured = mocker.register_uri('GET', self.sysconfig['http_url'],
+                            text='Foo Bar',
+                            additional_matcher=check_authorization_header)
+        log_line = "Either 'http_username' or 'http_password' are given, but for HTTP Authentication, both must be set\."
+        self.run_bot(parameters={'http_username': 'username'}, allowed_warning_count=1)
+        self.assertLogMatches(log_line, 'WARNING')
+        self.run_bot(parameters={'http_password': 'password'}, allowed_warning_count=1)
+        self.assertLogMatches(log_line, 'WARNING')
+
+    def test_auth(self, mocker):
+        """
+        Test the Authorization header when username and password are set.
+        """
+        def check_authorization_header(request):
+            print(f'check_authorization_header: {request.headers}')  # show the erroneos headers when the test fails
+            return request.headers['Authorization'] == 'Basic dXNlcjpwYXNzd29yZA=='
+        # generates a mock address if the Authorization header is correct, otherwise the test fails
+        captured = mocker.register_uri('GET', self.sysconfig['http_url'],
+                            text='Foo Bar',
+                            additional_matcher=check_authorization_header)
+        self.run_bot(parameters={'http_username': 'user',
+                                 'http_password': 'password'})
+
+
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()