Renew tokens via Firebase refresh token instead of replaying stored password

.gitignore

@@ -1,4 +1,6 @@
 .boostcamp/
+.venv/
+.pytest_cache/
 __pycache__/
 *.pyc
 *.egg-info/

BOOSTCAMP_API.md

@@ -26,9 +26,19 @@ The app uses the Firebase Identity Toolkit to exchange credentials for a token.
     "returnSecureToken": true
   }
   ```
-- **Response:** Returns `idToken`, which is used in the `Authorization` header.
+- **Response:** Returns `idToken` (used in the `Authorization` header) and a long-lived `refreshToken` (used to renew the ID token without re-sending credentials).
 
-#### 2. Request Password Reset (For OAuth Users)
+#### 2. Token Refresh
+Firebase ID tokens are short-lived (~1 hour). Instead of replaying the password, exchange the `refreshToken` at the secure-token endpoint.
+- **Endpoint:** `https://securetoken.googleapis.com/v1/token?key=<API_KEY>`
+- **Method:** `POST`
+- **Payload (form-encoded):**
+  ```
+  grant_type=refresh_token&refresh_token=<REFRESH_TOKEN>
+  ```
+- **Response:** Returns a fresh `id_token` and `refresh_token` (note the snake_case keys, unlike the login response).
+
+#### 3. Request Password Reset (For OAuth Users)
 - **Endpoint:** `https://www.googleapis.com/identitytoolkit/v3/relyingparty/getOobConfirmationCode?key=<API_KEY>`
 - **Method:** `POST`
 - **Payload:**

boostcampapi/boostcampapi.py

@@ -19,6 +19,7 @@ class BoostcampEndpoints(object):
     FIREBASE_API_KEY = "AIzaSyAEJcoGF-5ueF3bvaujcJm2PUV7RHKQwTw"
     FIREBASE_LOGIN_URL = f"https://www.googleapis.com/identitytoolkit/v3/relyingparty/verifyPassword?key={FIREBASE_API_KEY}"
     FIREBASE_RESET_URL = f"https://www.googleapis.com/identitytoolkit/v3/relyingparty/getOobConfirmationCode?key={FIREBASE_API_KEY}"
+    FIREBASE_REFRESH_URL = f"https://securetoken.googleapis.com/v1/token?key={FIREBASE_API_KEY}"
 
     @classmethod
     def get_user_endpoint(cls) -> str:
@@ -93,6 +94,7 @@ def __init__(
         self._session_file = session_file
         self._email: Optional[str] = None
         self._password: Optional[str] = None
+        self._refresh_token: Optional[str] = None
 
         self._headers = {
             "Accept": "*/*",
@@ -127,8 +129,9 @@ async def login(
     ) -> None:
         """Logs into Boostcamp using Firebase Identity Toolkit.
 
-        Stores credentials in the session file so that the client can
-        automatically re-authenticate when the token expires.
+        Captures the ID token and the long-lived refresh token. The refresh
+        token (not the password) is persisted to the session file so the
+        client can renew an expired ID token via the secure-token endpoint.
         """
         self._email = email
         self._password = password
@@ -152,6 +155,7 @@ async def login(
 
                     data = await response.json()
                     self.set_token(data["idToken"])
+                    self._refresh_token = data.get("refreshToken")
 
                     if save_session:
                         self.save_session()
@@ -174,22 +178,25 @@ async def request_password_reset(self, email: str) -> Dict[str, Any]:
                 return await response.json()
 
     def save_session(self, filename: Optional[str] = None) -> None:
-        """Saves the auth token and credentials to a pickle file."""
+        """Saves the auth token and refresh token to a pickle file.
+
+        The plaintext password is intentionally never persisted; token renewal
+        relies on the Firebase refresh token instead.
+        """
         if filename is None:
             filename = self._session_file
         filename = os.path.abspath(filename)
 
         session_data = {
             "token": self._token,
-            "email": self._email,
-            "password": self._password,
+            "refresh_token": self._refresh_token,
         }
         os.makedirs(os.path.dirname(filename), exist_ok=True)
         with open(filename, "wb") as fh:
             pickle.dump(session_data, fh)
 
     def load_session(self, filename: Optional[str] = None) -> bool:
-        """Loads auth token and credentials from a pickle file."""
+        """Loads the auth token and refresh token from a pickle file."""
         if filename is None:
             filename = self._session_file
 
@@ -199,10 +206,40 @@ def load_session(self, filename: Optional[str] = None) -> bool:
         with open(filename, "rb") as fh:
             data = pickle.load(fh)
             self.set_token(data["token"])
-            self._email = data.get("email")
-            self._password = data.get("password")
+            self._refresh_token = data.get("refresh_token")
             return True
 
+    async def _refresh_session(self) -> bool:
+        """Exchange the stored refresh token for a fresh ID token.
+
+        Uses the Firebase secure-token endpoint, which is the Firebase-native
+        way to renew a short-lived ID token without re-sending credentials.
+        Returns True on success, False if there is no refresh token or the
+        exchange fails.
+        """
+        if not self._refresh_token:
+            return False
+
+        payload = {
+            "grant_type": "refresh_token",
+            "refresh_token": self._refresh_token,
+        }
+        try:
+            async with ClientSession() as session:
+                async with session.post(
+                    BoostcampEndpoints.FIREBASE_REFRESH_URL,
+                    data=payload,
+                    timeout=self._timeout,
+                ) as response:
+                    if response.status != 200:
+                        return False
+                    data = await response.json()
+                    self.set_token(data["id_token"])
+                    self._refresh_token = data.get("refresh_token", self._refresh_token)
+                    return True
+        except Exception:
+            return False
+
     async def _re_login(self) -> bool:
         """Attempt to re-login using stored credentials."""
         if not self._email or not self._password:
@@ -232,7 +269,9 @@ async def _post(self, endpoint: str, data: Optional[Dict[str, Any]] = None) -> D
                     return await response.json()
             except ClientResponseError as e:
                 if e.status == 403:
-                    if await self._re_login():
+                    # Prefer the Firebase refresh-token exchange; fall back to a
+                    # password re-login only if credentials are still in memory.
+                    if await self._refresh_session() or await self._re_login():
                         async with ClientSession(headers=self._headers) as retry_session:
                             async with retry_session.post(
                                 endpoint,

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto

requirements-dev.txt

@@ -0,0 +1,3 @@
+-r requirements.txt
+pytest>=8.0
+pytest-asyncio>=0.23

tests/conftest.py

@@ -0,0 +1,84 @@
+"""Test helpers: a version-independent fake ``aiohttp.ClientSession``.
+
+aioresponses pins to aiohttp internals that drift between releases, so instead
+we patch the ``ClientSession`` symbol used by the client with a small fake that
+exercises the real request/response code paths (payload construction, status
+handling, token updates) without touching the network.
+"""
+from collections import defaultdict, deque
+from types import SimpleNamespace
+
+import pytest
+from aiohttp.client_exceptions import ClientResponseError
+
+import boostcampapi.boostcampapi as boostcampapi
+
+
+class _FakeResponse:
+    def __init__(self, status, json_data=None, text_data=""):
+        self.status = status
+        self._json = json_data
+        self._text = text_data
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def json(self):
+        return self._json
+
+    async def text(self):
+        return self._text
+
+    def raise_for_status(self):
+        if self.status >= 400:
+            raise ClientResponseError(
+                request_info=SimpleNamespace(real_url="http://test"),
+                history=(),
+                status=self.status,
+                message=f"HTTP {self.status}",
+            )
+
+
+class FakeHTTP:
+    """Registers queued responses per URL and records the calls made."""
+
+    def __init__(self):
+        self._responses = defaultdict(deque)
+        self.calls = []
+
+    def register(self, url, status=200, json=None, text=""):
+        self._responses[url].append(_FakeResponse(status, json, text))
+
+    def _session_factory(fake_http):
+        class _FakeSession:
+            def __init__(self, *args, headers=None, **kwargs):
+                self._headers = dict(headers) if headers else {}
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *exc):
+                return False
+
+            def post(self, url, json=None, data=None, timeout=None):
+                fake_http.calls.append(
+                    SimpleNamespace(
+                        url=url, json=json, data=data, headers=self._headers
+                    )
+                )
+                queue = fake_http._responses.get(url)
+                if not queue:
+                    raise AssertionError(f"No fake response registered for {url}")
+                return queue.popleft()
+
+        return _FakeSession
+
+
+@pytest.fixture
+def fake_http(monkeypatch):
+    fake = FakeHTTP()
+    monkeypatch.setattr(boostcampapi, "ClientSession", fake._session_factory())
+    return fake

tests/test_auth.py

@@ -0,0 +1,95 @@
+import pickle
+
+from boostcampapi.boostcampapi import BoostcampAPI, BoostcampEndpoints
+
+
+async def test_login_captures_refresh_token(fake_http):
+    """login() should keep the refreshToken from the Firebase response, not just the idToken."""
+    fake_http.register(
+        BoostcampEndpoints.FIREBASE_LOGIN_URL,
+        json={"idToken": "tok1", "refreshToken": "refresh1", "expiresIn": "3600"},
+    )
+    api = BoostcampAPI()
+    await api.login("user@example.com", "hunter2", save_session=False)
+
+    assert api.token == "tok1"
+    assert api._refresh_token == "refresh1"
+
+
+async def test_save_session_does_not_persist_password(fake_http, tmp_path):
+    """The pickled session must contain the refresh token but never the plaintext password."""
+    fake_http.register(
+        BoostcampEndpoints.FIREBASE_LOGIN_URL,
+        json={"idToken": "tok1", "refreshToken": "refresh1", "expiresIn": "3600"},
+    )
+    session_file = tmp_path / "session.pickle"
+    api = BoostcampAPI(session_file=str(session_file))
+    await api.login("user@example.com", "hunter2")
+
+    with open(session_file, "rb") as fh:
+        stored = pickle.load(fh)
+
+    assert stored["token"] == "tok1"
+    assert stored["refresh_token"] == "refresh1"
+    assert "password" not in stored
+    assert "hunter2" not in stored.values()
+
+
+def test_load_session_restores_refresh_token(tmp_path):
+    """load_session() should restore the token and refresh token from a saved session."""
+    session_file = tmp_path / "session.pickle"
+    with open(session_file, "wb") as fh:
+        pickle.dump({"token": "tokA", "refresh_token": "refreshA"}, fh)
+
+    api = BoostcampAPI(session_file=str(session_file))
+    assert api.load_session() is True
+    assert api.token == "tokA"
+    assert api._refresh_token == "refreshA"
+
+
+async def test_refresh_session_exchanges_refresh_token(fake_http):
+    """_refresh_session() should POST a refresh_token grant and adopt the new tokens."""
+    fake_http.register(
+        BoostcampEndpoints.FIREBASE_REFRESH_URL,
+        json={"id_token": "tok2", "refresh_token": "refresh2", "expires_in": "3600"},
+    )
+    api = BoostcampAPI(token="tok1")
+    api._refresh_token = "refresh1"
+
+    assert await api._refresh_session() is True
+    assert api.token == "tok2"
+    assert api._refresh_token == "refresh2"
+
+    call = fake_http.calls[-1]
+    assert call.url == BoostcampEndpoints.FIREBASE_REFRESH_URL
+    sent = call.data or call.json
+    assert sent["grant_type"] == "refresh_token"
+    assert sent["refresh_token"] == "refresh1"
+
+
+async def test_refresh_session_returns_false_without_refresh_token(fake_http):
+    """_refresh_session() should be a no-op (return False) when no refresh token is held."""
+    api = BoostcampAPI(token="tok1")
+    assert await api._refresh_session() is False
+    assert fake_http.calls == []
+
+
+async def test_post_refreshes_token_on_403_and_retries(fake_http):
+    """A 403 should trigger a refresh-token exchange and a retry with the new token."""
+    endpoint = BoostcampEndpoints.get_user_endpoint()
+    fake_http.register(endpoint, status=403)
+    fake_http.register(
+        BoostcampEndpoints.FIREBASE_REFRESH_URL,
+        json={"id_token": "tok2", "refresh_token": "refresh2", "expires_in": "3600"},
+    )
+    fake_http.register(endpoint, status=200, json={"user": "me"})
+
+    api = BoostcampAPI(token="tok1")
+    api._refresh_token = "refresh1"
+    result = await api.get_user_profile()
+
+    assert result == {"user": "me"}
+    assert api.token == "tok2"
+    retry_call = fake_http.calls[-1]
+    assert retry_call.url == endpoint
+    assert retry_call.headers["Authorization"] == "FirebaseIdToken:tok2"