casper/devdisc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v1

Browse Source
This commit is contained in:
casper
2026-06-08 06:17:48 +00:00
parent db1c6969a3
commit 7889828829
5 changed files with 4021 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

691
main.py Normal file
View File

@@ -0,0 +1,691 @@
# v6
from __future__ import annotations
import logging
import os
import sys
from logging.handlers import RotatingFileHandler
import traceback
from theme import (
apply_dark_palette,
apply_stylesheet,
)
from PySide6.QtCore import Qt
from PySide6.QtGui import QIcon
from PySide6.QtWidgets import (
QApplication,
QLabel,
QMainWindow,
QMessageBox,
QStatusBar,
)
# whats needed
from browser import BrowserView
from config import (
APP_NAME,
APP_VERSION,
DATABASE_PATH,
LOG_LEVEL,
LOGS_DIR,
START_URL,
WINDOW_HEIGHT,
WINDOW_WIDTH,
ensure_directories,
load_settings,
save_settings,
)
from db import Database
from managers import (
AddonManagerWindow,
NetworkInspector,
PluginManagerWindow,
ProfileManager,
SettingsWindow,
TrayManager,
ContextManager,
)
# v2
os.environ.setdefault(
"QTWEBENGINE_CHROMIUM_FLAGS",
(
"--disable-features="
"Translate,"
"MediaRouter "
"--enable-features="
"DnsOverHttps "
"--dns-over-https-mode=secure "
"--dns-over-https-templates="
"https://dns.quad9.net/dns-query "
"--disable-background-networking "
"--disable-component-update "
"--disable-domain-reliability "
"--disable-breakpad "
"--no-pings "
"--enable-gpu-rasterization "
"--enable-zero-copy "
"--disable-sync "
"--disable-translate "
"--disable-features=AutofillServerCommunication "
"--enable-native-gpu-memory-buffers "
),
)
LOG_FILE = LOGS_DIR / "client.log"
def configure_logging() -> None:
LOG_FILE.parent.mkdir(
parents=True,
exist_ok=True,
)
root = logging.getLogger()
root.setLevel(
LOG_LEVEL
)
root.handlers.clear()
formatter = logging.Formatter(
"%(asctime)s %(levelname)s %(name)s: %(message)s"
)
file_handler = (
RotatingFileHandler(
LOG_FILE,
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
)
file_handler.setFormatter(
formatter
)
console_handler = (
logging.StreamHandler()
)
console_handler.setFormatter(
formatter
)
root.addHandler(
file_handler
)
root.addHandler(
console_handler
)
configure_logging()
log = logging.getLogger(
__name__
)
class MainWindow(QMainWindow):
def __init__(
self,
db: Database,
settings: dict,
):
super().__init__()
self.db = db
self.settings_data = settings
self.contexts = ContextManager(
self.db,
self.settings_data,
)
self.plugins_window = None
self.settings_window = None
self.profile_window = None
self.addons_window = None
self.network_window = None
self.tray_manager = None
self.setWindowTitle(
f"{APP_NAME}---{APP_VERSION}"
)
self.resize(
max(
800,
int(
settings.get(
"window_width",
WINDOW_WIDTH,
)
),
),
max(
600,
int(
settings.get(
"window_height",
WINDOW_HEIGHT,
)
),
),
)
self.browser = BrowserView(
db=self.db,
context=self.context,
settings=self.settings_data,
parent=self,
)
self.setCentralWidget(
self.browser
)
self._build_status_bar()
self._connect_signals()
self._build_tray()
self.browser.open(
START_URL
)
@property
def context(self):
return self.contexts.current()
def _build_status_bar(
self,
) -> None:
self.status_bar = (
QStatusBar(
self
)
)
self.setStatusBar(
self.status_bar
)
self.download_label = QLabel(
"Ready",
self,
)
self.status_bar.addPermanentWidget(
self.download_label
)
self.status_bar.addPermanentWidget(
QLabel(
f"Blocked: "
f"{self.browser.network.blocklist_engine.count():,}"
)
)
def _connect_signals(
self,
) -> None:
self.browser.downloadProgress.connect(
self._download_progress
)
self.browser.page_.consoleMessage.connect(
self._console_message
)
self.browser.bridge.messageReceived.connect(
self._bridge_message
)
def _build_tray(
self,
) -> None:
tray_cfg = self.settings_data.get(
"tray",
{},
)
if not tray_cfg.get(
"enabled",
True,
):
return
self.tray_manager = (
TrayManager(
self
)
)
self.tray_manager.open_action.triggered.connect(
self._show_window
)
self.tray_manager.tray.show()
def _show_window(
self,
) -> None:
self.showNormal()
self.raise_()
self.activateWindow()
def _console_message(
self,
message: str,
) -> None:
ignored = {
"setMemoryInformation not available",
"[ProcessUtilsElectron]",
}
if any(
token in message
for token in ignored
):
return
log.info(
"console=%s",
message,
)
def _bridge_message(
self,
message: str,
) -> None:
log.info(
"bridge=%s",
message,
)
def _download_progress(
self,
filename: str,
received: int,
total: int,
) -> None:
if total <= 0:
self.download_label.setText(
f"{filename}: {received:,} B"
)
return
percent = int(
(
received
/ max(
total,
1,
)
)
* 100
)
self.download_label.setText(
(
f"{filename}: "
f"{percent}% "
f"({received:,}/{total:,})"
)
)
def open_settings(
self,
) -> None:
window = SettingsWindow(
self.settings_data,
self,
)
window.settingsChanged.connect(
self._settings_changed
)
self.settings_window = window
window.exec()
def open_plugins(
self,
) -> None:
self.plugins_window = (
PluginManagerWindow(
self.browser,
self.db,
self.context.name,
self,
)
)
self.plugins_window.show()
def open_profiles(
self,
) -> None:
window = ProfileManager(
self.db.list_profiles(),
self.settings_data.get(
"active_profile",
"default",
),
self,
)
window.profileSelected.connect(
self._profile_selected
)
self.profile_window = window
window.exec()
def open_addons(
self,
) -> None:
self.addons_window = (
AddonManagerWindow(
self.context,
self,
)
)
self.addons_window.show()
def open_network(
self,
) -> None:
self.network_window = (
NetworkInspector(
self.browser.network,
self,
)
)
self.network_window.show()
def open_devtools(
self,
) -> None:
self.browser.open_devtools()
def _settings_changed(
self,
) -> None:
save_settings(
self.settings_data
)
def _profile_selected(
self,
profile_name: str,
) -> None:
current = (
self.settings_data.get(
"active_profile"
)
)
if profile_name == current:
return
self.settings_data[
"active_profile"
] = profile_name
save_settings(
self.settings_data
)
QMessageBox.information(
self,
APP_NAME,
(
"Profile updated.\n\n"
"Restart required."
),
)
def _persist_window_state(
self,
) -> None:
size = self.size()
self.settings_data[
"window_width"
] = size.width()
self.settings_data[
"window_height"
] = size.height()
save_settings(
self.settings_data
)
def _shutdown(
self,
) -> None:
try:
self.browser.shutdown()
except Exception:
log.exception(
"browser shutdown failed"
)
try:
self.db.shutdown()
except Exception:
log.exception(
"database shutdown failed"
)
# v2
def closeEvent(
self,
event,
):
if getattr(
self,
"_closing",
False,
):
event.accept()
return
tray_cfg = self.settings_data.get(
"tray",
{},
)
if (
tray_cfg.get(
"close_to_tray",
True,
)
and self.tray_manager
):
event.ignore()
self.hide()
return
self._closing = True
try:
self._persist_window_state()
for widget in (
self.plugins_window,
self.settings_window,
self.profile_window,
self.addons_window,
self.network_window,
):
if widget:
try:
widget.close()
except Exception:
log.exception(
"child window close failed"
)
self._shutdown()
except Exception:
log.exception(
"shutdown failed"
)
event.accept()
super().closeEvent(
event
)
def build_database() -> Database:
return Database(
DATABASE_PATH
)
# v3
def build_app() -> QApplication:
os.environ.setdefault(
"QTWEBENGINE_REMOTE_DEBUGGING",
"0",
)
os.environ.setdefault(
"CHROME_CRASHPAD_PIPE_NAME",
"",
)
QApplication.setAttribute(
Qt.ApplicationAttribute.AA_ShareOpenGLContexts
)
return QApplication(
sys.argv
)
def fatal_error(
text: str,
) -> None:
try:
app = QApplication.instance()
if app is None:
app = QApplication(
sys.argv
)
QMessageBox.critical(
None,
APP_NAME,
text,
)
except Exception:
pass
# v3
def main() -> int:
try:
ensure_directories()
settings = load_settings()
db = build_database()
QApplication.setDesktopSettingsAware(
True
)
app = build_app()
icon = QIcon(
"app.ico"
)
if icon.isNull():
for candidate in (
"ui/app.ico",
"assets/app.ico",
"icon.ico",
):
test = QIcon(
candidate
)
if not test.isNull():
icon = test
break
if not icon.isNull():
app.setWindowIcon(
icon
)
app.setApplicationName(
APP_NAME
)
app.setApplicationVersion(
APP_VERSION
)
app.setStyle(
"Fusion"
)
apply_dark_palette(
app
)
app.setStyleSheet(
apply_stylesheet()
)
window = MainWindow(
db=db,
settings=settings,
)
window.show()
return app.exec()
except KeyboardInterrupt:
return 0
except SystemExit:
return 0
except Exception as exc:
log.exception(
"fatal startup error"
)
fatal_error(
str(exc)
)
return 1
if __name__ == "__main__":
raise SystemExit(
main()
)

1609
managers.py Normal file
View File

File diff suppressed because it is too large Load Diff

768
models.py Normal file
View File

@@ -0,0 +1,768 @@
# v7
from __future__ import annotations
from typing import Any
from dataclasses import dataclass
from pathlib import Path
from PySide6.QtCore import (
QAbstractTableModel,
QModelIndex,
Qt,
)
@dataclass(slots=True)
class ProfileContext:
name: str
root: Path
storage_path: Path
cache_path: Path
downloads_path: Path
plugins_path: Path
themes_path: Path
extensions_path: Path
profile_db: Path
permissions_db: Path
network_db: Path
class NetworkTableModel(
QAbstractTableModel
):
HEADERS = (
"Method",
"Status",
"Host",
"URL",
"Type",
"Duration",
"Timestamp",
)
def __init__(
self,
parent=None,
):
super().__init__(
parent
)
self._rows: list[
dict[str, Any]
] = []
def set_rows(
self,
rows,
) -> None:
self.beginResetModel()
self._rows = [
dict(row)
for row in rows
]
self.endResetModel()
def append_row(
self,
row: dict,
) -> None:
position = len(
self._rows
)
self.beginInsertRows(
QModelIndex(),
position,
position,
)
self._rows.append(
row
)
self.endInsertRows()
def rowCount(
self,
parent=QModelIndex(),
) -> int:
return 0 if parent.isValid() else len(
self._rows
)
def columnCount(
self,
parent=QModelIndex(),
) -> int:
return 0 if parent.isValid() else len(
self.HEADERS
)
def headerData(
self,
section,
orientation,
role,
):
if (
role
!= Qt.ItemDataRole.DisplayRole
):
return None
if (
orientation
== Qt.Orientation.Horizontal
):
if (
0
<= section
< len(
self.HEADERS
)
):
return self.HEADERS[
section
]
return str(
section + 1
)
def data(
self,
index,
role,
):
if (
not index.isValid()
or role
!= Qt.ItemDataRole.DisplayRole
):
return None
row = self._rows[
index.row()
]
match index.column():
case 0:
return row.get(
"method",
"",
)
case 1:
return row.get(
"status_code",
row.get(
"status",
"",
),
)
case 2:
return row.get(
"host",
"",
)
case 3:
return row.get(
"url",
"",
)
case 4:
return row.get(
"resource_type",
"",
)
case 5:
duration = row.get(
"duration_ms"
)
if duration is None:
return ""
try:
return (
f"{float(duration):.2f}"
)
except (
TypeError,
ValueError,
):
return ""
case 6:
return row.get(
"timestamp",
"",
)
return None
def row(
self,
position: int,
) -> dict:
return self._rows[
position
]
class ProfileModel(
QAbstractTableModel
):
HEADERS = (
"Profile",
)
def __init__(
self,
parent=None,
):
super().__init__(
parent
)
self._profiles: list[
str
] = []
def set_profiles(
self,
profiles: list[str],
):
self.beginResetModel()
self._profiles = sorted(
profiles
)
self.endResetModel()
def rowCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else len(
self._profiles
)
def columnCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else 1
def headerData(
self,
section,
orientation,
role,
):
if (
role
!= Qt.ItemDataRole.DisplayRole
):
return None
if (
orientation
== Qt.Orientation.Horizontal
):
return self.HEADERS[
section
]
return None
def data(
self,
index,
role,
):
if (
not index.isValid()
or role
!= Qt.ItemDataRole.DisplayRole
):
return None
return self._profiles[
index.row()
]
class AddonModel(
QAbstractTableModel
):
HEADERS = (
"Name",
"Enabled",
"Type",
)
def __init__(
self,
parent=None,
):
super().__init__(
parent
)
self._addons: list[
dict
] = []
def set_addons(
self,
addons: list[dict],
):
self.beginResetModel()
self._addons = list(
addons
)
self.endResetModel()
def rowCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else len(
self._addons
)
def columnCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else 3
def headerData(
self,
section,
orientation,
role,
):
if (
role
!= Qt.ItemDataRole.DisplayRole
):
return None
if (
orientation
== Qt.Orientation.Horizontal
):
return self.HEADERS[
section
]
return None
def data(
self,
index,
role,
):
if (
not index.isValid()
or role
!= Qt.ItemDataRole.DisplayRole
):
return None
addon = self._addons[
index.row()
]
match index.column():
case 0:
return addon.get(
"name",
"",
)
case 1:
return (
"Yes"
if addon.get(
"enabled",
True,
)
else "No"
)
case 2:
return addon.get(
"type",
"",
)
return None
class PluginTableModel(
QAbstractTableModel
):
HEADERS = (
"Enabled",
"Name",
"Version",
"Author",
"Status",
)
def __init__(
self,
parent=None,
):
super().__init__(
parent
)
self._plugins = []
def set_plugins(
self,
plugins,
):
self.beginResetModel()
self._plugins = [
dict(plugin)
for plugin in plugins
]
self.endResetModel()
def rowCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else len(
self._plugins
)
def columnCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else 5
def headerData(
self,
section,
orientation,
role,
):
if (
role
!= Qt.ItemDataRole.DisplayRole
):
return None
if (
orientation
== Qt.Orientation.Horizontal
):
return self.HEADERS[
section
]
return None
def data(
self,
index,
role,
):
if (
not index.isValid()
or role
!= Qt.ItemDataRole.DisplayRole
):
return None
plugin = self._plugins[
index.row()
]
match index.column():
case 0:
return (
"Yes"
if plugin.get(
"enabled",
True,
)
else "No"
)
case 1:
return plugin.get(
"name",
"",
)
case 2:
return plugin.get(
"version",
"",
)
case 3:
return plugin.get(
"author",
"",
)
case 4:
return plugin.get(
"status",
"Loaded",
)
return None
def row(
self,
position: int,
) -> dict:
return self._plugins[
position
]
class PluginLogTableModel(
QAbstractTableModel
):
HEADERS = (
"Time",
"Plugin",
"Level",
"Message",
)
def __init__(
self,
parent=None,
):
super().__init__(
parent
)
self._rows = []
def set_rows(
self,
rows,
):
self.beginResetModel()
self._rows = [
dict(row)
for row in rows
]
self.endResetModel()
def rowCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else len(
self._rows
)
def columnCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else 4
def headerData(
self,
section,
orientation,
role,
):
if (
role
!= Qt.ItemDataRole.DisplayRole
):
return None
if (
orientation
== Qt.Orientation.Horizontal
):
return self.HEADERS[
section
]
return None
def data(
self,
index,
role,
):
if (
not index.isValid()
or role
!= Qt.ItemDataRole.DisplayRole
):
return None
row = self._rows[
index.row()
]
match index.column():
case 0:
return row.get(
"created_at",
"",
)
case 1:
return row.get(
"plugin_name",
"",
)
case 2:
return row.get(
"level",
"",
)
case 3:
return row.get(
"message",
"",
)
return None
class PluginCrashTableModel(
QAbstractTableModel
):
HEADERS = (
"Time",
"Plugin",
"Error",
)
def __init__(
self,
parent=None,
):
super().__init__(
parent
)
self._rows = []
def set_rows(
self,
rows,
):
self.beginResetModel()
self._rows = [
dict(row)
for row in rows
]
self.endResetModel()
def rowCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else len(
self._rows
)
def columnCount(
self,
parent=QModelIndex(),
):
return 0 if parent.isValid() else 3
def headerData(
self,
section,
orientation,
role,
):
if (
role
!= Qt.ItemDataRole.DisplayRole
):
return None
if (
orientation
== Qt.Orientation.Horizontal
):
return self.HEADERS[
section
]
return None
def data(
self,
index,
role,
):
if (
not index.isValid()
or role
!= Qt.ItemDataRole.DisplayRole
):
return None
row = self._rows[
index.row()
]
match index.column():
case 0:
return row.get(
"created_at",
"",
)
case 1:
return row.get(
"plugin_name",
"",
)
case 2:
return row.get(
"error",
"",
)
return None

851
network.py Normal file
View File

@@ -0,0 +1,851 @@
from __future__ import annotations
import json
import logging
import socket
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from dataclasses import dataclass
from PySide6.QtCore import QObject, Signal
from config import DISCORD_TELEMETRY_BLOCKS
from blocklists import (
BlocklistEngine,
)
log = logging.getLogger(__name__)
REDACT_HEADERS = {
"authorization",
"cookie",
"set-cookie",
"x-super-properties",
"x-fingerprint",
"token",
"access_token",
"refresh_token",
}
DISCORD_GATEWAY_HOSTS = {
"gateway.discord.gg",
"remote-auth-gateway.discord.gg",
}
@dataclass(slots=True)
class ProxyConfig:
enabled: bool = False
proxy_type: str = "http"
host: str = ""
port: int = 0
username: str = ""
password: str = ""
@property
def valid(self) -> bool:
return (
self.enabled
and bool(self.host)
and self.port > 0
)
def to_dict(
self,
) -> dict[str, Any]:
return {
"enabled": self.enabled,
"proxy_type": self.proxy_type,
"host": self.host,
"port": self.port,
"username": self.username,
"password": self.password,
}
@classmethod
def from_dict(
cls,
data: dict[str, Any] | None,
) -> "ProxyConfig":
data = data or {}
return cls(
enabled=bool(
data.get(
"enabled",
False,
)
),
proxy_type=str(
data.get(
"proxy_type",
"http",
)
),
host=str(
data.get(
"host",
"",
)
),
port=int(
data.get(
"port",
0,
)
),
username=str(
data.get(
"username",
"",
)
),
password=str(
data.get(
"password",
"",
)
),
)
class ProxyManager(QObject):
proxyChanged = Signal(dict)
def __init__(
self,
db,
):
super().__init__()
self.db = db
def get_proxy(
self,
profile_name: str,
) -> ProxyConfig:
return ProxyConfig.from_dict(
self.db.get_setting(
f"proxy:{profile_name}",
{},
)
)
def save_proxy(
self,
profile_name: str,
proxy: ProxyConfig,
) -> None:
self.db.set_setting(
f"proxy:{profile_name}",
proxy.to_dict(),
)
self.proxyChanged.emit(
proxy.to_dict()
)
def clear_proxy(
self,
profile_name: str,
) -> None:
self.save_proxy(
profile_name,
ProxyConfig(),
)
def test_proxy(
self,
proxy: ProxyConfig,
timeout: float = 5.0,
) -> bool:
if not proxy.valid:
return False
try:
with socket.create_connection(
(
proxy.host,
proxy.port,
),
timeout=timeout,
):
return True
except Exception:
return False
# v2
class DomainBlocker:
DEFAULT_BLOCKS = set(
DISCORD_TELEMETRY_BLOCKS
)
def __init__(
self,
db,
blocklist_engine=None,
):
self.db = db
self.blocklist = (
blocklist_engine
)
def domains(
self,
) -> set[str]:
domains = set(
self.DEFAULT_BLOCKS
)
try:
domains.update(
self.db.blocked_domains()
)
except Exception:
log.exception(
"blocked domain load failed"
)
return {
str(domain)
.lower()
.strip()
for domain in domains
if domain
}
def add(
self,
domain: str,
) -> None:
domain = (
domain
.lower()
.strip()
)
if domain:
self.db.add_blocked_domain(
domain
)
def remove(
self,
domain: str,
) -> None:
try:
with self.db.connection() as conn:
conn.execute(
"""
DELETE FROM blocked_domains
WHERE domain=?
""",
(
domain.lower()
.strip(),
),
)
conn.commit()
except Exception:
log.exception(
"domain removal failed"
)
# v2
def is_blocked(
self,
host: str,
) -> bool:
host = (
host.lower()
.strip()
)
if (
self.blocklist
and self.blocklist.contains(
host
)
):
return True
for domain in self.domains():
if (
host == domain
or host.endswith(
f".{domain}"
)
):
return True
return False
class HeaderRules:
def __init__(
self,
db,
):
self.db = db
def all(
self,
) -> list[dict]:
return self.db.get_setting(
"header_rules",
[],
)
def save(
self,
rules: list[dict],
) -> None:
self.db.set_setting(
"header_rules",
rules,
)
@staticmethod
def redact(
headers: dict,
) -> dict:
result = {}
for key, value in (
headers or {}
).items():
if (
str(key)
.lower()
.strip()
in REDACT_HEADERS
):
result[key] = (
"[REDACTED]"
)
else:
result[key] = value
return result
def apply(
self,
headers: dict,
) -> dict:
result = dict(
headers or {}
)
for rule in self.all():
action = str(
rule.get(
"action",
"",
)
).lower()
name = str(
rule.get(
"name",
"",
)
)
value = rule.get(
"value",
"",
)
if not name:
continue
if action in {
"add",
"replace",
}:
result[name] = value
elif action == "remove":
result.pop(
name,
None,
)
return result
# v5
# v3
class NetworkRules:
def __init__(
self,
settings: dict,
):
self.settings = settings
def load(
self,
profile_name: str | None = None,
plugin_name: str | None = None,
) -> list[dict]:
try:
rules = self.settings.get(
"network_rules",
[],
)
if not isinstance(
rules,
list,
):
return []
except Exception:
log.exception(
"network rule load failed"
)
return []
output: list[dict] = []
for rule in rules:
if not isinstance(
rule,
dict,
):
continue
if (
profile_name
and rule.get(
"profile"
)
not in {
None,
profile_name,
}
):
continue
if (
plugin_name
and rule.get(
"plugin"
)
not in {
None,
plugin_name,
}
):
continue
output.append(
rule
)
return output
def evaluate(
self,
host: str,
profile_name: str | None = None,
plugin_name: str | None = None,
) -> tuple[bool, str | None]:
host = (
host.lower()
.strip()
)
rules = sorted(
self.load(
profile_name,
plugin_name,
),
key=lambda r: (
0
if r.get(
"plugin"
)
else (
1
if r.get(
"profile"
)
else 2
)
),
)
for rule in rules:
pattern = (
str(
rule.get(
"pattern",
"",
)
)
.lower()
.strip()
)
if not pattern:
continue
matched = (
host == pattern
or host.endswith(
f".{pattern}"
)
)
if not matched:
continue
action = (
str(
rule.get(
"action",
"",
)
)
.lower()
.strip()
)
if action in {
"block",
"deny",
}:
return (
False,
None,
)
if action == "allow":
return (
True,
None,
)
if action == "redirect":
target = (
str(
rule.get(
"target",
"",
)
)
.strip()
)
return (
True,
target or None,
)
return (
True,
None,
)
class RequestReplay:
def __init__(
self,
db,
):
self.db = db
def get_request(
self,
request_id: int,
):
with self.db.connection() as conn:
return conn.execute(
"""
SELECT *
FROM network_requests
WHERE id=?
""",
(
request_id,
),
).fetchone()
class HARExporter:
def __init__(
self,
db,
):
self.db = db
def export(
self,
output_file: str | Path,
limit: int = 5000,
) -> Path:
output_file = Path(
output_file
)
entries = []
for row in self.db.recent_requests(
limit
):
entries.append(
{
"startedDateTime": row[
"timestamp"
],
"request": {
"method": row[
"method"
],
"url": row[
"url"
],
},
"response": {
"status": row[
"status_code"
],
},
}
)
output_file.write_text(
json.dumps(
{
"log": {
"version": "1.2",
"creator": {
"name": "Discord Client",
"version": "1.0",
},
"entries": entries,
}
},
indent=2,
ensure_ascii=False,
),
encoding="utf-8",
)
return output_file
# v4# v5
class NetworkBackend(QObject):
requestAdded = Signal(dict)
def __init__(
self,
db,
settings: dict,
):
super().__init__()
self.db = db
self.settings = settings
mode = self.db.get_setting(
"blocklist_mode",
"pro",
)
blocklist_dir = Path(
"config/blocklists"
)
blocklists = sorted(
path
for path in blocklist_dir.glob(
"*.txt"
)
if path.is_file()
)
if mode == "pro":
blocklists = [
path
for path in blocklists
if "ultimate"
not in path.name.lower()
]
cache_file = (
blocklist_dir
/ f"{mode}.cache.bin"
)
self.blocklist_engine = (
BlocklistEngine(
blocklists=blocklists,
cache_file=cache_file,
)
)
try:
self.blocklist_engine.load()
except Exception:
log.exception(
"blocklist load failed"
)
self.blocker = DomainBlocker(
db,
self.blocklist_engine,
)
self.headers = HeaderRules(
db
)
self.rules = NetworkRules(
settings
)
self.proxy = ProxyManager(
db
)
self.har = HARExporter(
db
)
self.replay = RequestReplay(
db
)
def recent(
self,
limit: int = 1000,
):
return self.db.recent_requests(
limit
)
def search(
self,
query: str,
limit: int = 1000,
):
return self.db.search_requests(
query,
limit,
)
@staticmethod
def host_from_url(
url: str,
) -> str:
try:
return (
urlparse(url)
.hostname
or ""
).lower()
except Exception:
return ""
def process_request(
self,
request: dict,
) -> tuple[dict, bool]:
url = str(
request.get(
"url",
"",
)
)
host = self.host_from_url(
url
)
request["host"] = host
if host in DISCORD_GATEWAY_HOSTS:
self.requestAdded.emit(
request
)
return request, False
if (
host
and self.blocker.is_blocked(
host
)
):
request["_blocked"] = True
self.requestAdded.emit(
request
)
return request, True
allowed, redirect_url = (
self.rules.evaluate(
host=host,
profile_name=request.get(
"profile"
),
plugin_name=request.get(
"plugin"
),
)
)
if not allowed:
request["_blocked"] = True
self.requestAdded.emit(
request
)
return request, True
if redirect_url:
request[
"redirect_url"
] = redirect_url
headers = self.headers.apply(
request.get(
"headers",
{},
)
)
request["headers"] = (
self.headers.redact(
headers
)
)
self.requestAdded.emit(
request
)
return request, False

102
theme.py Normal file
View File

@@ -0,0 +1,102 @@
# v1
from PySide6.QtGui import QColor
from PySide6.QtGui import QPalette
from PySide6.QtWidgets import QApplication
def apply_dark_palette(
app: QApplication,
) -> None:
palette = QPalette()
palette.setColor(
QPalette.ColorRole.Window,
QColor(24, 24, 27),
)
palette.setColor(
QPalette.ColorRole.WindowText,
QColor(250, 250, 250),
)
palette.setColor(
QPalette.ColorRole.Base,
QColor(17, 17, 20),
)
palette.setColor(
QPalette.ColorRole.AlternateBase,
QColor(30, 30, 34),
)
palette.setColor(
QPalette.ColorRole.Text,
QColor(250, 250, 250),
)
palette.setColor(
QPalette.ColorRole.Button,
QColor(39, 39, 42),
)
palette.setColor(
QPalette.ColorRole.ButtonText,
QColor(250, 250, 250),
)
palette.setColor(
QPalette.ColorRole.Highlight,
QColor(59, 130, 246),
)
palette.setColor(
QPalette.ColorRole.HighlightedText,
QColor(255, 255, 255),
)
palette.setColor(
QPalette.ColorRole.ToolTipBase,
QColor(24, 24, 27),
)
palette.setColor(
QPalette.ColorRole.ToolTipText,
QColor(250, 250, 250),
)
app.setPalette(
palette
)
def apply_stylesheet(
) -> str:
return """
QPushButton {
border:none;
border-radius:6px;
padding:6px 12px;
}
QLineEdit,
QTextEdit,
QPlainTextEdit,
QComboBox,
QListWidget,
QTreeWidget,
QTableView {
border:1px solid #3f3f46;
border-radius:6px;
padding:4px;
}
QTabWidget::pane {
border:none;
}
QToolTip {
border:1px solid #3f3f46;
}
"""