feat(mcp): store OAuth — codes d'autorisation (#114)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -88,3 +88,50 @@ def get_client(db_path, client_id: str) -> dict | None:
|
|||||||
d = dict(row)
|
d = dict(row)
|
||||||
d["client_metadata"] = json.loads(d["client_metadata"])
|
d["client_metadata"] = json.loads(d["client_metadata"])
|
||||||
return d
|
return d
|
||||||
|
|
||||||
|
|
||||||
|
def save_code(
|
||||||
|
db_path,
|
||||||
|
code: str,
|
||||||
|
*,
|
||||||
|
client_id: str,
|
||||||
|
user_id: int,
|
||||||
|
redirect_uri: str | None,
|
||||||
|
code_challenge: str | None,
|
||||||
|
code_challenge_method: str | None,
|
||||||
|
scope: str | None,
|
||||||
|
resource: str | None,
|
||||||
|
expires_at: str,
|
||||||
|
) -> None:
|
||||||
|
with _connect(db_path) as conn:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO oauth_codes (code_hash, client_id, user_id, redirect_uri, "
|
||||||
|
"code_challenge, code_challenge_method, scope, resource, expires_at) "
|
||||||
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
|
(
|
||||||
|
_hash(code),
|
||||||
|
client_id,
|
||||||
|
user_id,
|
||||||
|
redirect_uri,
|
||||||
|
code_challenge,
|
||||||
|
code_challenge_method,
|
||||||
|
scope,
|
||||||
|
resource,
|
||||||
|
expires_at,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
|
||||||
|
|
||||||
|
def get_code(db_path, code: str) -> dict | None:
|
||||||
|
with _connect(db_path) as conn:
|
||||||
|
row = conn.execute(
|
||||||
|
"SELECT * FROM oauth_codes WHERE code_hash = ?", (_hash(code),)
|
||||||
|
).fetchone()
|
||||||
|
return dict(row) if row else None
|
||||||
|
|
||||||
|
|
||||||
|
def delete_code(db_path, code: str) -> None:
|
||||||
|
with _connect(db_path) as conn:
|
||||||
|
conn.execute("DELETE FROM oauth_codes WHERE code_hash = ?", (_hash(code),))
|
||||||
|
conn.commit()
|
||||||
|
|||||||
@@ -13,3 +13,26 @@ def test_create_and_get_client(tmp_path):
|
|||||||
assert row["client_id"] == "abc123"
|
assert row["client_id"] == "abc123"
|
||||||
assert row["client_metadata"] == meta
|
assert row["client_metadata"] == meta
|
||||||
assert store.get_client(db, "nope") is None
|
assert store.get_client(db, "nope") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_get_delete_code(tmp_path):
|
||||||
|
db = tmp_path / "u.sqlite"
|
||||||
|
store.init_schema(db)
|
||||||
|
store.save_code(
|
||||||
|
db,
|
||||||
|
"thecode",
|
||||||
|
client_id="abc",
|
||||||
|
user_id=7,
|
||||||
|
redirect_uri="https://claude.ai/api/mcp/auth_callback",
|
||||||
|
code_challenge="chal",
|
||||||
|
code_challenge_method="S256",
|
||||||
|
scope="mcp",
|
||||||
|
resource="https://colibre.fr/_mcp",
|
||||||
|
expires_at="2999-01-01T00:00:00+00:00",
|
||||||
|
)
|
||||||
|
row = store.get_code(db, "thecode")
|
||||||
|
assert row["user_id"] == 7
|
||||||
|
assert row["code_challenge"] == "chal"
|
||||||
|
assert row["resource"] == "https://colibre.fr/_mcp"
|
||||||
|
store.delete_code(db, "thecode")
|
||||||
|
assert store.get_code(db, "thecode") is None
|
||||||
|
|||||||
Reference in New Issue
Block a user