Merge pull request #23 from logzio/feature/support-dockerhub-audit-logs

[feature] support dockerhub audit logs

.github/workflows/unitTests.yaml

@@ -16,4 +16,4 @@ jobs:
           pip install pytest
           pip install pytest-cov
       - name: Run tests
-        run: PYTHONPATH=. pytest --cov-report term-missing --cov=src tests/UnitTests/test_*.py
+        run: PYTHONPATH=. pytest --cov-report term-missing --cov=src tests/UnitTests/test_*.py

README.md

@@ -230,6 +230,30 @@ By default `1password` API type has built in pagination settings and sets the `r
 | onepassword_limit        | 1Password limit for number of events to return in a single request (allowed range: 100 to 1000)              | Optional          | 100               |
 | pagination_off           | True if builtin pagination should be off, False otherwise                                                    | Optional          | `False`           |
 
+</details>
+
+<details>
+  <summary>
+    <span><a href="./src/apis/dockerhub/README.md">Dockerhub</a></span>
+  </summary>
+
+For dockerhub audit logs, use type `dockerhub` with the below parameters.
+
+## Configuration Options
+| Parameter Name         | Description                                                                           | Required/Optional | Default           |
+|------------------------|---------------------------------------------------------------------------------------|-------------------|-------------------|
+| name                   | Name of the API (custom name)                                                         | Optional          | the defined `url` |
+| dockerhub_user         | DockerHub username                                                                    | Required          | -                 |
+| dockerhub_token        | DockerHub personal access token or password                                           | Required          | -                 |
+| url                    | The request URL                                                                       | Required          | -                 |
+| next_url               | URL for the next page of results (used for pagination)                                | Optional          | -                 |
+| method                 | The request method (`GET` or `POST`)                                                  | Optional          | `GET`             |
+| days_back_fetch        | Number of days to fetch back in the first request. Adds a filter on `from` parameter. | Optional          | -1                |
+| refresh_token_interval | Interval in minutes to refresh the JWT token                                          | Optional          | 30 (minute)       |
+| scrape_interval        | Time interval to wait between runs (unit: `minutes`)                                  | Optional          | 1 (minute)        |
+| additional_fields      | Additional custom fields to add to the logs before sending to logzio                  | Optional          | -                 |
+
+
 </details>
 
 
@@ -285,6 +309,8 @@ docker stop -t 30 logzio-api-fetcher
 ```
 
 ## Changelog:
+- **0.3.0**:
+  - Add support for dockerhub audit logs
 - **0.2.2**:
   - Resolve Azure mail reports bug
 - **0.2.1**:

src/apis/dockerhub/Dockerhub.py

@@ -0,0 +1,74 @@
+import time
+from datetime import datetime, timedelta, UTC
+import json
+import logging
+import requests
+from pydantic import Field
+from src.apis.general.Api import ApiFetcher
+
+logger = logging.getLogger(__name__)
+
+
+class DockerHub(ApiFetcher):
+    """
+    :param dockerhub_user: The DockerHub username
+    :param dockerhub_token: The DockerHub personal access token or password
+    :param days_back_fetch: Number of days to fetch back in the first request, Optional (adds a filter on 'from')
+    :param page_size: Number of events to return in a single request (for pagination)
+    :param refresh_token_interval: Interval in minutes to refresh the JWT token
+    """
+    dockerhub_user: str = Field(frozen=True)
+    dockerhub_token: str = Field(frozen=False)
+    days_back_fetch: int = Field(default=-1, frozen=True)
+    refresh_token_interval: int = Field(default=30)
+    _jwt_token: str = None
+    _token_expiry: datetime = None
+
+    def __init__(self, **data):
+        res_data_path = "logs"
+        headers = {
+            "Content-Type": "application/json",
+        }
+        super().__init__(headers=headers, response_data_path=res_data_path, **data)
+        self._initialize_params()
+
+    def _initialize_params(self):
+        try:
+            params = {"page_size": 100}
+            if self.days_back_fetch > 0:
+                from_date = (datetime.now(UTC) - timedelta(days=self.days_back_fetch)).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+                params["from"] = from_date
+            query_string = "&".join([f"{key}={value}" for key, value in params.items()])
+            if "?" in self.url:
+                self.url += f"&{query_string}"
+            else:
+                self.url += f"?{query_string}"
+        except Exception as e:
+            logger.error(
+                f"Failed to update request params. Sending {self.name} request with default params. Error: {e}")
+
+    def _get_jwt_token(self):
+        if self._jwt_token and datetime.now(UTC) < self._token_expiry:
+            return self._jwt_token
+
+        url = "https://hub.docker.com/v2/users/login"
+        payload = {
+            "username": self.dockerhub_user,
+            "password": self.dockerhub_token
+        }
+        headers = {"Content-Type": "application/json"}
+        try:
+            response = requests.post(url, json=payload, headers=headers)
+            response.raise_for_status()
+            token_response = response.json()
+            self._jwt_token = token_response.get("token")
+            self._token_expiry = datetime.now(UTC) + timedelta(minutes=self.refresh_token_interval)
+            return self._jwt_token
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Failed to get JWT token: {e}")
+
+    def send_request(self):
+        session_token = self._get_jwt_token()
+        self.headers["Authorization"] = f"Bearer {session_token}"
+        response = super().send_request()
+        return response

src/apis/dockerhub/README.md

@@ -0,0 +1,42 @@
+# DockerHub API Configuration
+The `dockerhub` API type is used to fetch audit logs from DockerHub. It supports pagination and allows filtering logs based on a date range.
+
+## Configuration
+| Parameter Name         | Description                                                                           | Required/Optional | Default           |
+|------------------------|---------------------------------------------------------------------------------------|-------------------|-------------------|
+| name                   | Name of the API (custom name)                                                         | Optional          | the defined `url` |
+| dockerhub_user         | DockerHub username                                                                    | Required          | -                 |
+| dockerhub_token        | DockerHub personal access token or password                                           | Required          | -                 |
+| url                    | The request URL                                                                       | Required          | -                 |
+| next_url               | URL for the next page of results (used for pagination)                                | Optional          | -                 |
+| method                 | The request method (`GET` or `POST`)                                                  | Optional          | `GET`             |
+| days_back_fetch        | Number of days to fetch back in the first request. Adds a filter on `from` parameter. | Optional          | -1                |
+| refresh_token_interval | Interval in minutes to refresh the JWT token                                          | Optional          | 30 (minute)       |
+| scrape_interval        | Time interval to wait between runs (unit: `minutes`)                                  | Optional          | 1 (minute)        |
+| additional_fields      | Additional custom fields to add to the logs before sending to logzio                  | Optional          | -                 |
+
+## Example
+You can customize the endpoints to collect data from by adding extra API configurations under `apis`. DockerHub API Docs can be found [here](https://docs.docker.com/docker-hub/api/latest/).
+
+Example configuration:
+
+```yaml
+apis:
+  - name: Dockerhub audit logs
+    type: dockerhub
+    dockerhub_token: <<docker_hub_password>>
+    dockerhub_user: <<docker_hub_username>>
+    url: https://hub.docker.com/v2/auditlogs/<<dockerhub_account>>
+    next_url: https://hub.docker.com/v2/auditlogs/logzio?from={res.logs.[0].timestamp}
+    method: GET
+    days_back_fetch: 7
+    scrape_interval: 1
+    refresh_token_interval: 20
+    additional_fields:
+      type: dockerhub-audit
+      eventType: auditevents
+
+logzio:
+  url: https://<<LISTENER-HOST>>:8071
+  token: <<LOG-SHIPPING-TOKEN>>
+```

src/apis/dockerhub/test_e2e_dockerhub.py

@@ -0,0 +1,101 @@
+import glob
+import json
+import os
+import threading
+from os.path import abspath, dirname
+import requests
+import unittest
+import yaml
+
+from src.main import main
+
+TEST_TYPE = "dockerhub-audit-test"
+
+def _search_data(query):
+    """
+    Send given search query to logzio and returns the result.
+    :param query:
+    :return:
+    """
+    url = "https://api.logz.io/v1/search"
+    headers = {
+        "X-API-TOKEN": os.environ["LOGZIO_API_TOKEN"],
+        "CONTENT-TYPE": "application/json",
+        "ACCEPT": "application/json"
+    }
+    body = {
+        "query": {
+            "query_string": {
+                "query": query
+            }
+        }
+    }
+
+    r = requests.post(url=url, headers=headers, json=body)
+    if r:
+        data = json.loads(r.text)
+        hits = data.get("hits").get("hits")
+        return hits
+    return []
+
+
+def delete_temp_files():
+    """
+    delete the temp config that generated for the test
+    """
+    curr_path = abspath(dirname(dirname(__file__)))
+    test_configs_path = f"{curr_path}/testdata/*_temp.yaml"
+
+    for file in glob.glob(test_configs_path):
+        os.remove(file)
+
+
+def _update_config_tokens(file_path):
+    """
+    Updates the tokens in the given file.
+    """
+    with open(file_path, "r") as conf:
+        content = yaml.safe_load(conf)
+    e = os.environ
+    if "DOCKERHUB_TOKEN" not in os.environ:
+        raise EnvironmentError("DOCKERHUB_TOKEN environment variable is missing")
+    content["apis"][0]["dockerhub_token"] = os.environ["DOCKERHUB_TOKEN"]
+
+    if "DOCKERHUB_USER" not in os.environ:
+        raise EnvironmentError("DOCKERHUB_USER environment variable is missing")
+    content["apis"][0]["dockerhub_user"] = os.environ["DOCKERHUB_USER"]
+
+    if "LOGZIO_SHIPPING_TOKEN" not in os.environ:
+        raise EnvironmentError("LOGZIO_SHIPPING_TOKEN environment variable is missing")
+    content["logzio"]["token"] = os.environ["LOGZIO_SHIPPING_TOKEN"]
+
+    path, ext = file_path.rsplit(".", 1)
+    temp_test_path = f"{path}_temp.{ext}"
+
+    with open(temp_test_path, "w") as file:
+        yaml.dump(content, file)
+
+    return temp_test_path
+
+
+class TestDockerhubE2E(unittest.TestCase):
+    """
+    Test data arrived to logzio
+    """
+
+    def test_data_in_logz(self):
+        curr_path = abspath(dirname(__file__))
+        config_path = f"{curr_path}/testdata/valid_dockerhub_config.yaml"
+        temp_config_path = _update_config_tokens(config_path)
+        thread = threading.Thread(target=main, kwargs={"conf_path": temp_config_path})
+        thread.daemon = True
+        thread.start()
+        thread.join(timeout=60)
+        azure_logs_in_acc = _search_data(f"type:{TEST_TYPE}")
+        self.assertTrue(azure_logs_in_acc)
+        self.assertTrue(all([log.get("_source").get("eventType") == "auditevents" for log in azure_logs_in_acc]))
+        delete_temp_files()
+
+
+if __name__ == '__main__':
+    unittest.main()

src/apis/dockerhub/testdata/valid_dockerhub_config.yaml

@@ -0,0 +1,20 @@
+apis:
+  - name: Dockerhub audit logs
+    type: dockerhub
+    dockerhub_token: token
+    dockerhub_user: user
+    url: https://hub.docker.com/v2/auditlogs/logzio
+    next_url: https://hub.docker.com/v2/auditlogs/logzio?from={res.logs.[0].timestamp}
+    method: GET
+    days_back_fetch: 7
+    scrape_interval: 10
+    additional_fields:
+      type: dockerhub-audit-test
+      eventType: auditevents
+
+logzio:
+  url: https://listener.logz.io:8071
+  token: token
+
+
+

src/apis/dockerhub/utils/logging_config.ini

@@ -0,0 +1,22 @@
+[loggers]
+keys = root
+
+[handlers]
+keys = stream_handler
+
+[formatters]
+keys = formatter
+
+[logger_root]
+level = INFO
+handlers = stream_handler
+
+[handler_stream_handler]
+class = StreamHandler
+level = INFO
+formatter = formatter
+args = (sys.stderr,)
+
+[formatter_formatter]
+class=src.utils.MaskInfoFormatter.MaskInfoFormatter
+format = %(asctime)s [%(levelname)s]: %(message)s

src/config/ConfigReader.py

@@ -11,6 +11,7 @@
 from src.apis.cloudflare.Cloudflare import Cloudflare
 from src.apis.onepassword.OnePassword import OnePassword
 from src.output.LogzioShipper import LogzioShipper
+from src.apis.dockerhub.Dockerhub import DockerHub
 
 INPUT_API_FIELD = "apis"
 OUTPUT_LOGZIO_FIELD = "logzio"
@@ -21,7 +22,8 @@
     "azure_graph": "AzureGraph",
     "azure_mail_reports": "AzureMailReports",
     "cloudflare": "Cloudflare",
-    "1password": "OnePassword"
+    "1password": "OnePassword",
+    "dockerhub": "DockerHub"
 }
 
 logger = logging.getLogger(__name__)