From 6bd8c564f1773ea395aed634489a1bfe71c11897 Mon Sep 17 00:00:00 2001 From: Framewrk CI Date: Mon, 13 Jul 2026 17:40:12 +0200 Subject: [PATCH] Harden credential and session security --- README.md | 22 ++++++++++++- pytr/account.py | 40 ++++++++++++++---------- pytr/api.py | 23 ++++++++------ tests/test_security.py | 70 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 129 insertions(+), 26 deletions(-) create mode 100644 tests/test_security.py diff --git a/README.md b/README.md index adcdb8d9..11f48698 100644 --- a/README.md +++ b/README.md @@ -138,9 +138,29 @@ Install dependencies: Run the tests to ensure everything is set up correctly: ```sh - uv run pytest +uv run pytest ``` +### Security hardening in this fork + +This fork changes authentication storage and logging to reduce the impact of a +local file, shell-history, or debug-log disclosure: + +- The PIN is never written to disk. `--store_credentials` stores only the + phone number and the authenticated session cookies. +- Existing legacy two-line credential files are automatically reduced to the + phone number; a PIN is requested interactively if the saved session expires. +- Credential and cookie files are created with owner-only permissions (`0600`). +- Debug logging no longer dumps login responses, account settings, websocket + payloads, or websocket errors, which can contain sensitive account data. +- The PIN should be entered interactively rather than passed as `--pin`, so it + does not enter shell history or appear in process arguments. + +These changes address plaintext PIN persistence and accidental disclosure of +authentication/session/account data through logs. The SDK still communicates +with Trade Republic's private API and stores session cookies locally; protect +the `~/.pytr` directory and avoid enabling debug logs in shared environments. + ### Linting and Code Formatting This project uses [Ruff](https://astral.sh/ruff) for code linting and auto-formatting, as well as diff --git a/pytr/account.py b/pytr/account.py index 5f688547..07676844 100644 --- a/pytr/account.py +++ b/pytr/account.py @@ -1,4 +1,5 @@ import json +import os import sys import time from getpass import getpass @@ -24,38 +25,45 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh If no parameters are set but are needed then ask for input """ log = get_logger(__name__) - save_cookies = True + using_stored_phone = phone_no is None and CREDENTIALS_FILE.is_file() + save_cookies = using_stored_phone or store_credentials if phone_no is None and CREDENTIALS_FILE.is_file(): with open(CREDENTIALS_FILE) as f: lines = f.readlines() phone_no = lines[0].strip() - pin = lines[1].strip() + if len(lines) > 1: + fd = os.open(CREDENTIALS_FILE, os.O_WRONLY | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + f.write(phone_no + "\n") + os.chmod(CREDENTIALS_FILE, 0o600) + log.info("Removed the legacy stored PIN from the credentials file.") phone_no_masked = phone_no[:-8] + "********" - pin_masked = len(pin) * "*" - log.info(f"Using credentials from file {CREDENTIALS_FILE}. Phone: {phone_no_masked}, PIN: {pin_masked}") + log.info(f"Using phone number from file {CREDENTIALS_FILE}: {phone_no_masked}") else: BASE_DIR.mkdir(parents=True, exist_ok=True) if phone_no is None: print("Please enter your TradeRepublic phone number in the format +4912345678:") phone_no = input() - if pin is None: - print("Please enter your TradeRepublic pin:") - pin = getpass(prompt="Pin (Input is hidden):") - - if store_credentials: - with open(CREDENTIALS_FILE, "w") as f: - f.writelines([phone_no + "\n", pin + "\n"]) - - log.info(f"Storing credentials/cookies in {BASE_DIR}") - else: - save_cookies = False + if save_cookies: + BASE_DIR.mkdir(parents=True, exist_ok=True) tr = TradeRepublicApi(phone_no=phone_no, pin=pin, save_cookies=save_cookies, waf_token=waf_token) # Use same login as app.traderepublic.com if not tr.resume_websession(): + if pin is None: + print("Please enter your TradeRepublic pin:") + pin = getpass(prompt="Pin (Input is hidden):") + tr.pin = pin + + if store_credentials: + fd = os.open(CREDENTIALS_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + f.write(phone_no + "\n") + os.chmod(CREDENTIALS_FILE, 0o600) + log.info(f"Storing phone number and session cookies in {BASE_DIR}; PIN is not stored") try: countdown = tr.initiate_weblogin() except ValueError as e: @@ -79,5 +87,5 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh tr.complete_weblogin(code) log.info("Logged in.") - log.debug(get_settings(tr)) + log.debug("Account settings retrieved.") return tr diff --git a/pytr/api.py b/pytr/api.py index 521009be..b4f7242d 100644 --- a/pytr/api.py +++ b/pytr/api.py @@ -22,6 +22,7 @@ import asyncio import json +import os import pathlib import re import ssl @@ -83,17 +84,17 @@ def __init__( self._credentials_file = pathlib.Path(credentials_file) if credentials_file else CREDENTIALS_FILE - if not (phone_no and pin): + if not phone_no: try: with open(self._credentials_file, "r") as f: lines = f.readlines() self.phone_no = lines[0].strip() - self.pin = lines[1].strip() except FileNotFoundError: - raise ValueError(f"phone_no and pin must be specified explicitly or via {self._credentials_file}") + raise ValueError(f"phone_no must be specified explicitly or via {self._credentials_file}") else: self.phone_no = phone_no - self.pin = pin + + self.pin = pin self._cookies_file = pathlib.Path(cookies_file) if cookies_file else BASE_DIR / f"cookies.{self.phone_no}.txt" @@ -208,6 +209,9 @@ def _set_waf_cookie(self, token: str): def initiate_weblogin(self): self.log.info("Initiating web login...") + if not self.pin: + raise ValueError("pin must be specified explicitly; it is never read from or written to disk") + if self._waf_token == "awswaf": self._waf_token = self._fetch_waf_token_awswaf() elif self._waf_token == "playwright": @@ -228,7 +232,7 @@ def initiate_weblogin(self): self.log.debug(f"Web login returned: {r.status_code}") r.raise_for_status() j = r.json() - self.log.debug(f"Web login data: {json.dumps(j, indent=4)}") + self.log.debug("Web login response received.") try: self._process_id = j["processId"] except KeyError: @@ -258,6 +262,7 @@ def save_websession(self): if self._save_cookies: # Saves session cookies too (expirydate=0). self._websession.cookies.save(ignore_discard=True) + os.chmod(self._cookies_file, 0o600) def resume_websession(self): """ @@ -350,7 +355,7 @@ async def _next_subscription_id(self): async def subscribe(self, payload): subscription_id = await self._next_subscription_id() ws = await self._get_ws() - self.log.debug(f"Subscribing: 'sub {subscription_id} {json.dumps(payload)}'") + self.log.debug("Subscribing to request %s.", subscription_id) self.subscriptions[subscription_id] = payload await ws.send(f"sub {subscription_id} {json.dumps(payload)}") return subscription_id @@ -368,7 +373,7 @@ async def recv(self): ws = await self._get_ws() while True: response = await ws.recv() - self.log.debug(f"Received message: {response!r}") + self.log.debug("Received websocket response.") subscription_id = response[: response.find(" ")] code = response[response.find(" ") + 1 : response.find(" ") + 2] @@ -387,7 +392,7 @@ async def recv(self): elif code == "D": response = self._calculate_delta(subscription_id, payload_str) - self.log.debug(f"Payload is {response}") + self.log.debug("Received websocket delta response.") self._previous_responses[subscription_id] = response return subscription_id, subscription, json.loads(response) @@ -398,7 +403,7 @@ async def recv(self): continue elif code == "E": - self.log.error(f"Received error message: {response!r}") + self.log.error("Trade Republic returned a websocket error.") await self.unsubscribe(subscription_id) diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 00000000..1a8153f7 --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,70 @@ +import builtins +import logging +import os + +from pytr import account + + +def test_login_does_not_store_pin(monkeypatch, tmp_path): + credentials_file = tmp_path / "credentials" + monkeypatch.setattr(account, "BASE_DIR", tmp_path) + monkeypatch.setattr(account, "CREDENTIALS_FILE", credentials_file) + monkeypatch.setattr(account, "TradeRepublicApi", lambda **kwargs: FakeApi(**kwargs)) + monkeypatch.setattr(account, "getpass", lambda prompt: "1234") + monkeypatch.setattr(builtins, "input", lambda prompt="": "123456") + monkeypatch.setattr(account, "time", FakeTime) + + account.login(phone_no="+4912345678", store_credentials=True, waf_token="token") + + assert credentials_file.read_text() == "+4912345678\n" + assert credentials_file.read_text().splitlines() == ["+4912345678"] + assert os.stat(credentials_file).st_mode & 0o777 == 0o600 + + +def test_legacy_credentials_are_migrated(monkeypatch, tmp_path): + credentials_file = tmp_path / "credentials" + credentials_file.write_text("+4912345678\n1234\n") + monkeypatch.setattr(account, "BASE_DIR", tmp_path) + monkeypatch.setattr(account, "CREDENTIALS_FILE", credentials_file) + monkeypatch.setattr(account, "TradeRepublicApi", lambda **kwargs: FakeApi(**kwargs)) + monkeypatch.setattr(account, "getpass", lambda prompt: "5678") + monkeypatch.setattr(builtins, "input", lambda prompt="": "123456") + monkeypatch.setattr(account, "time", FakeTime) + + account.login(waf_token="token") + + assert credentials_file.read_text() == "+4912345678\n" + + +def test_debug_log_does_not_include_response_payload(caplog): + logger = logging.getLogger("security-test") + logger.setLevel(logging.DEBUG) + with caplog.at_level(logging.DEBUG, logger="security-test"): + logger.debug("Web login response received.") + + assert "1234" not in caplog.text + assert "pin" not in caplog.text.lower() + + +class FakeApi: + def __init__(self, **kwargs): + self.pin = kwargs["pin"] + + def resume_websession(self): + return False + + def initiate_weblogin(self): + return 0 + + def complete_weblogin(self, code): + assert self.pin == "1234" or self.pin == "5678" + + +class FakeTime: + @staticmethod + def time(): + return 0 + + @staticmethod + def sleep(seconds): + return None