Harden credential storage and sensitive logging

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

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

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)
 

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