1  """Tests for the project-secrets vault used by the Agentic Browser.
  2  
  3  Covers the round-trip through the REST API: create → read (masked) → patch
  4  (plaintext preserved when the sentinel comes back) → delete. Also spot-
  5  checks that `DBWrapper.resolve_project_secret` actually decrypts what was
  6  stored — the live-browser bit obviously isn't runnable here.
  7  """
  8  import pytest
  9  from fastapi.testclient import TestClient
 10  
 11  from restai.config import RESTAI_DEFAULT_PASSWORD
 12  from restai.database import get_db_wrapper
 13  from restai.main import app
 14  
 15  
 16  AUTH = ("admin", RESTAI_DEFAULT_PASSWORD)
 17  
 18  
 19  @pytest.fixture(scope="module")
 20  def client():
 21      with TestClient(app) as c:
 22          yield c
 23  
 24  
 25  @pytest.fixture
 26  def project_id(client):
 27      # Pick any project the admin has; create one if none.
 28      r = client.get("/projects", auth=AUTH)
 29      if r.status_code == 200:
 30          projects = r.json().get("projects", []) or []
 31          if projects:
 32              return projects[0]["id"]
 33  
 34      # Need a team + llm to create. Try to find one.
 35      teams = client.get("/teams", auth=AUTH).json().get("teams", []) or []
 36      team_id = teams[0]["id"] if teams else 1
 37  
 38      resp = client.post(
 39          "/projects",
 40          json={
 41              "name": "secret_test_project",
 42              "type": "agent",
 43              "llm": "fake",
 44              "team_id": team_id,
 45          },
 46          auth=AUTH,
 47      )
 48      if resp.status_code not in (200, 201):
 49          pytest.skip(f"Could not bootstrap a project for secret tests ({resp.status_code})")
 50      return resp.json()["id"]
 51  
 52  
 53  def test_secret_crud_roundtrip(client, project_id):
 54      # Create
 55      r = client.post(
 56          f"/projects/{project_id}/secrets",
 57          json={"name": "test_api_key", "value": "super-secret-plaintext", "description": "unit test"},
 58          auth=AUTH,
 59      )
 60      if r.status_code == 409:
 61          # Leftover from a previous run — delete + retry.
 62          existing = client.get(f"/projects/{project_id}/secrets", auth=AUTH).json()
 63          for s in existing:
 64              if s["name"] == "test_api_key":
 65                  client.delete(f"/projects/{project_id}/secrets/{s['id']}", auth=AUTH)
 66          r = client.post(
 67              f"/projects/{project_id}/secrets",
 68              json={"name": "test_api_key", "value": "super-secret-plaintext", "description": "unit test"},
 69              auth=AUTH,
 70          )
 71      assert r.status_code == 201, r.text
 72      created = r.json()
 73      assert created["name"] == "test_api_key"
 74      assert created["value"] == "********", "value must be masked on the response"
 75  
 76      secret_id = created["id"]
 77      try:
 78          # List — value still masked
 79          lst = client.get(f"/projects/{project_id}/secrets", auth=AUTH)
 80          assert lst.status_code == 200
 81          names = {s["name"]: s for s in lst.json()}
 82          assert "test_api_key" in names
 83          assert names["test_api_key"]["value"] == "********"
 84  
 85          # Resolve via DB helper — plaintext must match the original
 86          db = get_db_wrapper()
 87          try:
 88              plaintext = db.resolve_project_secret(project_id, "test_api_key")
 89          finally:
 90              db.db.close()
 91          assert plaintext == "super-secret-plaintext"
 92  
 93          # PATCH with masked value — plaintext must be preserved
 94          r = client.patch(
 95              f"/projects/{project_id}/secrets/{secret_id}",
 96              json={"value": "********", "description": "touched description"},
 97              auth=AUTH,
 98          )
 99          assert r.status_code == 200, r.text
100          db = get_db_wrapper()
101          try:
102              assert db.resolve_project_secret(project_id, "test_api_key") == "super-secret-plaintext"
103          finally:
104              db.db.close()
105  
106          # PATCH with a real new value — plaintext changes
107          r = client.patch(
108              f"/projects/{project_id}/secrets/{secret_id}",
109              json={"value": "rotated-value"},
110              auth=AUTH,
111          )
112          assert r.status_code == 200
113          db = get_db_wrapper()
114          try:
115              assert db.resolve_project_secret(project_id, "test_api_key") == "rotated-value"
116          finally:
117              db.db.close()
118  
119          # Duplicate name → 409
120          dup = client.post(
121              f"/projects/{project_id}/secrets",
122              json={"name": "test_api_key", "value": "x"},
123              auth=AUTH,
124          )
125          assert dup.status_code == 409
126      finally:
127          # Cleanup
128          client.delete(f"/projects/{project_id}/secrets/{secret_id}", auth=AUTH)
129  
130  
131  def test_resolve_missing_secret_returns_none(client, project_id):
132      db = get_db_wrapper()
133      try:
134          assert db.resolve_project_secret(project_id, "definitely_does_not_exist_xyz") is None
135      finally:
136          db.db.close()
137  
138  
139  def test_secret_name_validation(client, project_id):
140      # Slashes / spaces etc. must reject (validate_safe_name).
141      bad_names = ["has/slash", "has space", "../traversal"]
142      for name in bad_names:
143          r = client.post(
144              f"/projects/{project_id}/secrets",
145              json={"name": name, "value": "x"},
146              auth=AUTH,
147          )
148          assert r.status_code == 422, f"expected 422 for name {name!r}, got {r.status_code}"