updated to use .env, then pull idToken and clientID

scrape_costco.py

@@ -216,12 +216,13 @@ ITEM_FIELDS = [
 ]
 
 COSTCO_STORAGE_ORIGIN = "costco.com"
-COSTCO_AUTH_STORAGE_KEY = "costco-x-authorization"
-COSTCO_HEADERS_BLOB_KEY = "headers"
+COSTCO_ID_TOKEN_STORAGE_KEY = "idToken"
+COSTCO_CLIENT_ID_STORAGE_KEY = "clientID"
 
 def load_config():
     load_dotenv()
     return {
+        "authorization": os.getenv("COSTCO_X_AUTHORIZATION", "").strip(),
         "client_id": os.getenv("COSTCO_X_WCS_CLIENTID", "").strip(),
         "client_identifier": os.getenv("COSTCO_CLIENT_IDENTIFIER", "").strip(),
     }
@@ -244,33 +245,31 @@ def build_headers(auth_headers):
     return headers
 
 
-def load_costco_browser_headers(profile_dir, client_id, client_identifier):
+def load_costco_browser_headers(profile_dir, authorization, client_id, client_identifier):
     local_storage = read_firefox_local_storage(profile_dir, COSTCO_STORAGE_ORIGIN)
     webapps_store = read_firefox_webapps_store(profile_dir, COSTCO_STORAGE_ORIGIN)
-    auth_token = (
-        local_storage.get(COSTCO_AUTH_STORAGE_KEY, "").strip()
-        or webapps_store.get(COSTCO_AUTH_STORAGE_KEY, "").strip()
+    auth_header = authorization.strip() if authorization else ""
+    if client_id:
+        client_id = client_id.strip()
+    if client_identifier:
+        client_identifier = client_identifier.strip()
+
+    if not auth_header:
+        id_token = (
+            local_storage.get(COSTCO_ID_TOKEN_STORAGE_KEY, "").strip()
+            or webapps_store.get(COSTCO_ID_TOKEN_STORAGE_KEY, "").strip()
+        )
+        if id_token:
+            auth_header = f"Bearer {id_token}"
+
+    client_id = client_id or (
+        local_storage.get(COSTCO_CLIENT_ID_STORAGE_KEY, "").strip()
+        or webapps_store.get(COSTCO_CLIENT_ID_STORAGE_KEY, "").strip()
     )
 
-    if not auth_token:
-        header_blob = (
-            local_storage.get(COSTCO_HEADERS_BLOB_KEY, "").strip()
-            or webapps_store.get(COSTCO_HEADERS_BLOB_KEY, "").strip()
-        )
-        if header_blob:
-            try:
-                blob_data = json.loads(header_blob)
-            except json.JSONDecodeError:
-                blob_data = {}
-            auth_token = str(blob_data.get(COSTCO_AUTH_STORAGE_KEY, "")).strip()
-            client_id = client_id or str(blob_data.get("costco-x-wcs-clientId", "")).strip()
-            client_identifier = client_identifier or str(
-                blob_data.get("client-identifier", "")
-            ).strip()
-
-    if not auth_token:
+    if not auth_header:
         raise click.ClickException(
-            "could not find Costco auth token in Firefox session storage"
+            "could not find Costco auth token; set COSTCO_X_AUTHORIZATION or load Firefox idToken"
         )
     if not client_id or not client_identifier:
         raise click.ClickException(
@@ -278,7 +277,7 @@ def load_costco_browser_headers(profile_dir, client_id, client_identifier):
         )
 
     return {
-        "costco-x-authorization": auth_token,
+        "costco-x-authorization": auth_header,
         "costco-x-wcs-clientId": client_id,
         "client-identifier": client_identifier,
     }
@@ -666,6 +665,7 @@ def main(
 
     auth_headers = load_costco_browser_headers(
         profile_dir,
+        authorization=config["authorization"],
         client_id=config["client_id"],
         client_identifier=config["client_identifier"],
     )

tests/test_browser_session.py

@@ -33,7 +33,7 @@ class BrowserSessionTests(unittest.TestCase):
                 values["costco-x-wcs-clientId"],
             )
 
-    def test_load_costco_browser_headers_reads_exact_auth_key(self):
+    def test_load_costco_browser_headers_reads_id_token_and_client_id(self):
         with tempfile.TemporaryDirectory() as tmpdir:
             profile_dir = Path(tmpdir)
             storage_dir = profile_dir / "storage" / "default" / "https+++www.costco.com" / "ls"
@@ -44,12 +44,17 @@ class BrowserSessionTests(unittest.TestCase):
                 connection.execute("CREATE TABLE data (key TEXT, value TEXT)")
                 connection.execute(
                     "INSERT INTO data (key, value) VALUES (?, ?)",
-                    ("costco-x-authorization", "Bearer header.payload.signature"),
+                    ("idToken", "header.payload.signature"),
+                )
+                connection.execute(
+                    "INSERT INTO data (key, value) VALUES (?, ?)",
+                    ("clientID", "4900eb1f-0c10-4bd9-99c3-c59e6c1ecebf"),
                 )
 
             headers = scrape_costco.load_costco_browser_headers(
                 profile_dir,
-                client_id="4900eb1f-0c10-4bd9-99c3-c59e6c1ecebf",
+                authorization="",
+                client_id="",
                 client_identifier="481b1aec-aa3b-454b-b81b-48187e28f205",
             )
 
@@ -63,7 +68,7 @@ class BrowserSessionTests(unittest.TestCase):
             headers["client-identifier"],
         )
 
-    def test_load_costco_browser_headers_falls_back_to_exact_header_blob(self):
+    def test_load_costco_browser_headers_prefers_env_values(self):
         with tempfile.TemporaryDirectory() as tmpdir:
             profile_dir = Path(tmpdir)
             storage_dir = profile_dir / "storage" / "default" / "https+++www.costco.com" / "ls"
@@ -74,19 +79,22 @@ class BrowserSessionTests(unittest.TestCase):
                 connection.execute("CREATE TABLE data (key TEXT, value TEXT)")
                 connection.execute(
                     "INSERT INTO data (key, value) VALUES (?, ?)",
-                    (
-                        "headers",
-                        '{"costco-x-authorization":"Bearer header.payload.signature"}',
-                    ),
+                    ("idToken", "storage.payload.signature"),
+                )
+                connection.execute(
+                    "INSERT INTO data (key, value) VALUES (?, ?)",
+                    ("clientID", "4900eb1f-0c10-4bd9-99c3-c59e6c1ecebf"),
                 )
 
             headers = scrape_costco.load_costco_browser_headers(
                 profile_dir,
-                client_id="4900eb1f-0c10-4bd9-99c3-c59e6c1ecebf",
+                authorization="Bearer env.payload.signature",
+                client_id="env-client-id",
                 client_identifier="481b1aec-aa3b-454b-b81b-48187e28f205",
             )
 
-        self.assertEqual("Bearer header.payload.signature", headers["costco-x-authorization"])
+        self.assertEqual("Bearer env.payload.signature", headers["costco-x-authorization"])
+        self.assertEqual("env-client-id", headers["costco-x-wcs-clientId"])
 
     def test_scrape_costco_prompts_for_profile_dir_when_autodiscovery_fails(self):
         with mock.patch.object(
@@ -101,6 +109,7 @@ class BrowserSessionTests(unittest.TestCase):
             scrape_costco,
             "load_config",
             return_value={
+                "authorization": "",
                 "client_id": "4900eb1f-0c10-4bd9-99c3-c59e6c1ecebf",
                 "client_identifier": "481b1aec-aa3b-454b-b81b-48187e28f205",
             },

tests/test_costco_pipeline.py

@@ -414,6 +414,7 @@ class CostcoPipelineTests(unittest.TestCase):
                 scrape_costco,
                 "load_config",
                 return_value={
+                    "authorization": "",
                     "client_id": "4900eb1f-0c10-4bd9-99c3-c59e6c1ecebf",
                     "client_identifier": "481b1aec-aa3b-454b-b81b-48187e28f205",
                 },