digitalshadows/ds_lookup_username_connector.py at main · splunk-soar-connectors/digitalshadows · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
# Copyright (c) 2025-2026 Splunk Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# File: ds_lookup_username_connector.py
#
# Licensed under Apache 2.0 (https://www.apache.org/licenses/LICENSE-2.0.txt)
#

import phantom.app as phantom
from phantom.action_result import ActionResult

from digital_shadows_consts import DS_API_KEY_CFG, DS_API_SECRET_KEY_CFG, DS_LOOKUP_USERNAME_NOT_FOUND, DS_LOOKUP_USERNAME_SUCCESS
from dsapi.service.data_breach_username_service import DataBreachUsernameService


class DSLookupUsernameConnector:
    def __init__(self, connector):
        """
        :param connector: DigitalShadowsConnector
        """
        self._connector = connector

        config = connector.get_config()
        self._ds_api_key = config[DS_API_KEY_CFG]
        self._ds_api_secret_key = config[DS_API_SECRET_KEY_CFG]

    def lookup_username(self, param):
        action_result = ActionResult(dict(param))
        self._connector.add_action_result(action_result)

        username_to_lookup = param.get("username")

        username_service = DataBreachUsernameService(self._ds_api_key, self._ds_api_secret_key)
        view = username_service.data_breach_username_view(username=username_to_lookup)

        found_username = next(
            (
                data_breach_username
                for data_breach_username in username_service.find_all(view=view)
                if data_breach_username.username == username_to_lookup
            ),
            None,
        )

        if found_username is not None:
            summary, data = self._lookup_success(username_to_lookup, found_username)
            action_result.update_summary(summary)
            action_result.add_data(data)
            action_result.set_status(phantom.APP_SUCCESS, DS_LOOKUP_USERNAME_SUCCESS)
        else:
            summary, data = self._lookup_failure(username_to_lookup)
            action_result.update_summary(summary)
            action_result.add_data(data)
            action_result.set_status(phantom.APP_SUCCESS, DS_LOOKUP_USERNAME_NOT_FOUND)

        return action_result.get_status()

    def _lookup_success(self, username_queried, found_username):
        """
        Returns summary and data dictionaries for a successful username lookup.

        :param username_queried: str
        :param found_username: DataBreachUsernameSummary
        :return: tuple
        """
        summary = {"username_queried": username_queried, "username_was_found": True}
        data = {
            "username_retrieved": found_username.username,
            "username_was_found": True,
            "distinct_password_count": found_username.distinct_password_count,
            "breach_count": found_username.breach_count,
        }
        return summary, data

    def _lookup_failure(self, username_queried):
        """
        Returns summary and data dictionaries for an unsuccessful username lookup.

        :param username_queried: str
        :return: tuple
        """
        summary = {"username_queried": username_queried, "username_was_found": False}
        data = {"username_retrieved": "", "username_was_found": False, "distinct_password_count": 0, "breach_count": 0}
        return summary, data