admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-30 19:02:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add proper Google OAuth flow with localhost callback server

Browse Source 
Add exchange_google_token() and run_oauth_flow() to oauth.py for the
full authorization code exchange. Update gdrive, gcalendar, and
gcontacts connectors to trigger the browser-based OAuth flow when a
client_id:client_secret pair is provided, prefer access_token over raw
token, and require an actual access_token for is_connected(). auth_url()
now returns the Cloud Console credentials page when no client_id is
stored.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jon Saad-Falcon
2026-03-27 16:45:03 -07:00
co-authored by Claude Opus 4.6
parent 8d9eed6533
commit 62b98d903b
8 changed files with 452 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/connectors/gcalendar.py
+40 -8
View File
@@ -17,6 +17,7 @@ from openjarvis.connectors.oauth import (
build_google_auth_url,
delete_tokens,
load_tokens,
run_oauth_flow,
save_tokens,
)
from openjarvis.core.config import DEFAULT_CONFIG_DIR
@@ -214,11 +215,12 @@ class GCalendarConnector(BaseConnector):
# ------------------------------------------------------------------
def is_connected(self) -> bool:
"""Return ``True`` if a credentials file with a valid token exists."""
"""Return ``True`` if a credentials file with a valid access token exists."""
tokens = load_tokens(self._credentials_path)
if tokens is None:
return False
return bool(tokens)
# Must have an actual access_token, not just a client_id
return bool(tokens.get("access_token") or tokens.get("token"))
def disconnect(self) -> None:
"""Delete the stored credentials file."""
@@ -226,18 +228,48 @@ class GCalendarConnector(BaseConnector):
def auth_url(self) -> str:
"""Return a Google OAuth consent URL requesting ``calendar.readonly`` scope."""
tokens = load_tokens(self._credentials_path)
client_id = ""
if tokens:
client_id = tokens.get("client_id", "")
if not client_id:
return "https://console.cloud.google.com/apis/credentials"
return build_google_auth_url(
client_id="", # placeholder — real client_id from config
client_id=client_id,
scopes=[_GCAL_SCOPE],
)
def handle_callback(self, code: str) -> None:
"""Handle the OAuth callback by persisting the authorization code.
"""Handle the OAuth callback.
In a full implementation this would exchange the code for tokens.
For now the code is saved directly as the token value.
If *code* looks like a ``client_id:client_secret`` pair (containing
``.apps.googleusercontent.com``), store the credentials and trigger
the full browser-based OAuth flow. Otherwise treat it as a raw
token / auth code.
"""
save_tokens(self._credentials_path, {"token": code})
code = code.strip()
# If user pastes client_id:client_secret, store and run OAuth flow
if ":" in code and ".apps.googleusercontent.com" in code:
client_id, client_secret = code.split(":", 1)
try:
run_oauth_flow(
client_id=client_id.strip(),
client_secret=client_secret.strip(),
scopes=[_GCAL_SCOPE],
credentials_path=self._credentials_path,
)
except Exception: # noqa: BLE001
# Fallback: just save the credentials for later
save_tokens(
self._credentials_path,
{
"client_id": client_id.strip(),
"client_secret": client_secret.strip(),
},
)
else:
# Raw token or auth code
save_tokens(self._credentials_path, {"token": code})
def sync(
self,
@@ -261,7 +293,7 @@ class GCalendarConnector(BaseConnector):
if not tokens:
return
token: str = tokens.get("token", tokens.get("access_token", ""))
token: str = tokens.get("access_token", tokens.get("token", ""))
if not token:
return
src/openjarvis/connectors/gcontacts.py
+40 -10
View File
@@ -17,6 +17,7 @@ from openjarvis.connectors.oauth import (
build_google_auth_url,
delete_tokens,
load_tokens,
run_oauth_flow,
save_tokens,
)
from openjarvis.core.config import DEFAULT_CONFIG_DIR
@@ -161,13 +162,12 @@ class GContactsConnector(BaseConnector):
# ------------------------------------------------------------------
def is_connected(self) -> bool:
"""Return ``True`` if a credentials file with a valid token exists."""
"""Return ``True`` if a credentials file with a valid access token exists."""
tokens = load_tokens(self._credentials_path)
if tokens is None:
return False
# Accept any non-empty dict that contains at least one key
# (simplified: real impl would also check expiry / refresh token)
return bool(tokens)
# Must have an actual access_token, not just a client_id
return bool(tokens.get("access_token") or tokens.get("token"))
def disconnect(self) -> None:
"""Delete the stored credentials file."""
@@ -175,18 +175,48 @@ class GContactsConnector(BaseConnector):
def auth_url(self) -> str:
"""Return a Google OAuth consent URL requesting ``contacts.readonly`` scope."""
tokens = load_tokens(self._credentials_path)
client_id = ""
if tokens:
client_id = tokens.get("client_id", "")
if not client_id:
return "https://console.cloud.google.com/apis/credentials"
return build_google_auth_url(
client_id="", # placeholder — real client_id from config
client_id=client_id,
scopes=[_GCONTACTS_SCOPE],
)
def handle_callback(self, code: str) -> None:
"""Handle the OAuth callback by persisting the authorization code.
"""Handle the OAuth callback.
In a full implementation this would exchange the code for tokens.
For now the code is saved directly as the token value.
If *code* looks like a ``client_id:client_secret`` pair (containing
``.apps.googleusercontent.com``), store the credentials and trigger
the full browser-based OAuth flow. Otherwise treat it as a raw
token / auth code.
"""
save_tokens(self._credentials_path, {"token": code})
code = code.strip()
# If user pastes client_id:client_secret, store and run OAuth flow
if ":" in code and ".apps.googleusercontent.com" in code:
client_id, client_secret = code.split(":", 1)
try:
run_oauth_flow(
client_id=client_id.strip(),
client_secret=client_secret.strip(),
scopes=[_GCONTACTS_SCOPE],
credentials_path=self._credentials_path,
)
except Exception: # noqa: BLE001
# Fallback: just save the credentials for later
save_tokens(
self._credentials_path,
{
"client_id": client_id.strip(),
"client_secret": client_secret.strip(),
},
)
else:
# Raw token or auth code
save_tokens(self._credentials_path, {"token": code})
def sync(
self,
@@ -211,7 +241,7 @@ class GContactsConnector(BaseConnector):
if not tokens:
return
token: str = tokens.get("token", tokens.get("access_token", ""))
token: str = tokens.get("access_token", tokens.get("token", ""))
if not token:
return
src/openjarvis/connectors/gdrive.py
+40 -10
View File
@@ -17,6 +17,7 @@ from openjarvis.connectors.oauth import (
build_google_auth_url,
delete_tokens,
load_tokens,
run_oauth_flow,
save_tokens,
)
from openjarvis.core.config import DEFAULT_CONFIG_DIR
@@ -144,13 +145,12 @@ class GDriveConnector(BaseConnector):
# ------------------------------------------------------------------
def is_connected(self) -> bool:
"""Return ``True`` if a credentials file with a valid token exists."""
"""Return ``True`` if a credentials file with a valid access token exists."""
tokens = load_tokens(self._credentials_path)
if tokens is None:
return False
# Accept any non-empty dict that contains at least one key
# (simplified: real impl would also check expiry / refresh token)
return bool(tokens)
# Must have an actual access_token, not just a client_id
return bool(tokens.get("access_token") or tokens.get("token"))
def disconnect(self) -> None:
"""Delete the stored credentials file."""
@@ -158,18 +158,48 @@ class GDriveConnector(BaseConnector):
def auth_url(self) -> str:
"""Return a Google OAuth consent URL requesting ``drive.readonly`` scope."""
tokens = load_tokens(self._credentials_path)
client_id = ""
if tokens:
client_id = tokens.get("client_id", "")
if not client_id:
return "https://console.cloud.google.com/apis/credentials"
return build_google_auth_url(
client_id="", # placeholder — real client_id from config
client_id=client_id,
scopes=[_GDRIVE_SCOPE],
)
def handle_callback(self, code: str) -> None:
"""Handle the OAuth callback by persisting the authorization code.
"""Handle the OAuth callback.
In a full implementation this would exchange the code for tokens.
For now the code is saved directly as the token value.
If *code* looks like a ``client_id:client_secret`` pair (containing
``.apps.googleusercontent.com``), store the credentials and trigger
the full browser-based OAuth flow. Otherwise treat it as a raw
token / auth code.
"""
save_tokens(self._credentials_path, {"token": code})
code = code.strip()
# If user pastes client_id:client_secret, store and run OAuth flow
if ":" in code and ".apps.googleusercontent.com" in code:
client_id, client_secret = code.split(":", 1)
try:
run_oauth_flow(
client_id=client_id.strip(),
client_secret=client_secret.strip(),
scopes=[_GDRIVE_SCOPE],
credentials_path=self._credentials_path,
)
except Exception: # noqa: BLE001
# Fallback: just save the credentials for later
save_tokens(
self._credentials_path,
{
"client_id": client_id.strip(),
"client_secret": client_secret.strip(),
},
)
else:
# Raw token or auth code
save_tokens(self._credentials_path, {"token": code})
def sync(
self,
@@ -194,7 +224,7 @@ class GDriveConnector(BaseConnector):
if not tokens:
return
token: str = tokens.get("token", tokens.get("access_token", ""))
token: str = tokens.get("access_token", tokens.get("token", ""))
if not token:
return
src/openjarvis/connectors/oauth.py
+178
View File
@@ -95,3 +95,181 @@ def delete_tokens(path: str) -> None:
p = Path(path)
if p.exists():
p.unlink()
# ---------------------------------------------------------------------------
# Token exchange & full OAuth flow
# ---------------------------------------------------------------------------
def exchange_google_token(
code: str,
client_id: str,
client_secret: str,
redirect_uri: str = _DEFAULT_REDIRECT_URI,
) -> Dict[str, Any]:
"""Exchange an authorization code for access + refresh tokens.
Parameters
----------
code:
The authorization code received from Google's consent redirect.
client_id:
OAuth 2.0 client ID.
client_secret:
OAuth 2.0 client secret.
redirect_uri:
Must match the redirect URI used when obtaining the auth code.
Returns
-------
dict
Token response containing ``access_token``, ``refresh_token``,
``token_type``, and ``expires_in``.
"""
import httpx
resp = httpx.post(
"https://oauth2.googleapis.com/token",
data={
"code": code,
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": redirect_uri,
"grant_type": "authorization_code",
},
timeout=30.0,
)
resp.raise_for_status()
return resp.json()
def run_oauth_flow(
client_id: str,
client_secret: str,
scopes: List[str],
credentials_path: str,
redirect_uri: str = _DEFAULT_REDIRECT_URI,
) -> Dict[str, Any]:
"""Run the full OAuth flow: browser consent, callback, token exchange.
Steps:
1. Build consent URL
2. Start localhost callback server
3. Open browser to consent URL
4. Wait for Google to redirect with ``?code=...``
5. Exchange code for ``access_token`` + ``refresh_token``
6. Save tokens to *credentials_path*
7. Return the tokens dict
Parameters
----------
client_id:
OAuth 2.0 client ID.
client_secret:
OAuth 2.0 client secret.
scopes:
List of OAuth scopes to request.
credentials_path:
Where to persist the resulting tokens.
redirect_uri:
Local callback URI. Defaults to ``http://localhost:8789/callback``.
Returns
-------
dict
Token response from Google (``access_token``, ``refresh_token``, etc.).
Raises
------
RuntimeError
If the user denies authorization or the callback times out.
"""
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
auth_url = build_google_auth_url(
client_id=client_id,
redirect_uri=redirect_uri,
scopes=scopes,
)
# Mutable containers used by the callback handler closure.
auth_code: List[str] = []
error: List[str] = []
class _CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802 — required override name
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
if "code" in params:
auth_code.append(params["code"][0])
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(
b"<html><body><h2>Authorization successful!</h2>"
b"<p>You can close this tab and return to OpenJarvis.</p>"
b"</body></html>"
)
elif "error" in params:
error.append(params["error"][0])
self.send_response(400)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(
b"<html><body><h2>Authorization failed</h2>"
b"<p>Please try again.</p></body></html>"
)
else:
self.send_response(400)
self.end_headers()
def log_message(self, format: str, *args: Any) -> None: # noqa: A002
pass # Suppress HTTP request logs
# Parse port from redirect_uri
port = int(urlparse(redirect_uri).port or 8789)
server = HTTPServer(("127.0.0.1", port), _CallbackHandler)
server.timeout = 120 # 2 minute timeout
# Open the consent page in the user's default browser
webbrowser.open(auth_url)
# Wait for the callback (blocking, with per-request timeout)
while not auth_code and not error:
server.handle_request()
server.server_close()
if error:
raise RuntimeError(f"OAuth authorization failed: {error[0]}")
if not auth_code:
raise RuntimeError("OAuth authorization timed out")
# Exchange the authorization code for tokens
tokens = exchange_google_token(
code=auth_code[0],
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
)
# Persist tokens together with client credentials (needed for refresh)
save_tokens(
credentials_path,
{
"access_token": tokens.get("access_token", ""),
"refresh_token": tokens.get("refresh_token", ""),
"token_type": tokens.get("token_type", "Bearer"),
"expires_in": tokens.get("expires_in", 3600),
"client_id": client_id,
"client_secret": client_secret,
},
)
return tokens
tests/connectors/test_gcalendar.py
+3 -3
View File
@@ -79,11 +79,11 @@ def test_not_connected(connector) -> None:
def test_auth_url(connector) -> None:
"""auth_url() returns a URL to Google's OAuth endpoint with calendar scope."""
"""auth_url() returns the credentials page when no client_id is stored."""
url = connector.auth_url()
assert isinstance(url, str)
assert url.startswith("https://accounts.google.com/o/oauth2/v2/auth")
assert "calendar.readonly" in url
# Without a stored client_id, points to the Cloud Console credentials page
assert url == "https://console.cloud.google.com/apis/credentials"
# ---------------------------------------------------------------------------
tests/connectors/test_gcontacts.py
+3 -3
View File
@@ -77,11 +77,11 @@ def test_not_connected(connector) -> None:
def test_auth_url(connector) -> None:
"""auth_url() returns a URL to Google's OAuth endpoint with contacts scope."""
"""auth_url() returns the credentials page when no client_id is stored."""
url = connector.auth_url()
assert isinstance(url, str)
assert url.startswith("https://accounts.google.com/o/oauth2/v2/auth")
assert "contacts.readonly" in url
# Without a stored client_id, points to the Cloud Console credentials page
assert url == "https://console.cloud.google.com/apis/credentials"
# ---------------------------------------------------------------------------
tests/connectors/test_gdrive.py
+3 -3
View File
@@ -74,11 +74,11 @@ def test_not_connected(connector) -> None:
def test_auth_url(connector) -> None:
"""auth_url() returns a Google OAuth URL with drive.readonly scope."""
"""auth_url() returns the credentials page when no client_id is stored."""
url = connector.auth_url()
assert isinstance(url, str)
assert url.startswith("https://accounts.google.com/o/oauth2/v2/auth")
assert "drive.readonly" in url
# Without a stored client_id, points to the Cloud Console credentials page
assert url == "https://console.cloud.google.com/apis/credentials"
# ---------------------------------------------------------------------------
tests/connectors/test_oauth_flow.py
+145
View File
@@ -0,0 +1,145 @@
"""Tests for OAuth token exchange and Google connector integration."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
def test_exchange_google_token_calls_endpoint() -> None:
from openjarvis.connectors.oauth import exchange_google_token
mock_resp = MagicMock()
mock_resp.json.return_value = {
"access_token": "ya29.test",
"refresh_token": "1//test",
"token_type": "Bearer",
"expires_in": 3600,
}
mock_resp.raise_for_status = MagicMock()
with patch("httpx.post", return_value=mock_resp) as mock_post:
tokens = exchange_google_token(
code="4/test-code",
client_id="test-id.apps.googleusercontent.com",
client_secret="test-secret",
)
assert tokens["access_token"] == "ya29.test"
assert tokens["refresh_token"] == "1//test"
mock_post.assert_called_once()
def test_gdrive_handle_callback_triggers_oauth(tmp_path: Path) -> None:
from openjarvis.connectors.gdrive import GDriveConnector
creds = str(tmp_path / "gdrive.json")
conn = GDriveConnector(credentials_path=creds)
with patch("openjarvis.connectors.gdrive.run_oauth_flow") as mock_flow:
mock_flow.return_value = {"access_token": "ya29.test"}
conn.handle_callback("test-id.apps.googleusercontent.com:test-secret")
mock_flow.assert_called_once()
call_kwargs = mock_flow.call_args
assert "test-id.apps.googleusercontent.com" in str(call_kwargs)
def test_gdrive_is_connected_requires_access_token(tmp_path: Path) -> None:
from openjarvis.connectors.gdrive import GDriveConnector
from openjarvis.connectors.oauth import save_tokens
creds = str(tmp_path / "gdrive.json")
conn = GDriveConnector(credentials_path=creds)
# Just client_id is not "connected"
save_tokens(creds, {"client_id": "test-id"})
assert conn.is_connected() is False
# With access_token IS connected
save_tokens(creds, {"access_token": "ya29.test", "client_id": "test-id"})
assert conn.is_connected() is True
def test_gcalendar_handle_callback_triggers_oauth(tmp_path: Path) -> None:
from openjarvis.connectors.gcalendar import GCalendarConnector
creds = str(tmp_path / "gcalendar.json")
conn = GCalendarConnector(credentials_path=creds)
with patch("openjarvis.connectors.gcalendar.run_oauth_flow") as mock_flow:
mock_flow.return_value = {"access_token": "ya29.test"}
conn.handle_callback("test-id.apps.googleusercontent.com:test-secret")
mock_flow.assert_called_once()
def test_gcontacts_handle_callback_triggers_oauth(tmp_path: Path) -> None:
from openjarvis.connectors.gcontacts import GContactsConnector
creds = str(tmp_path / "gcontacts.json")
conn = GContactsConnector(credentials_path=creds)
with patch("openjarvis.connectors.gcontacts.run_oauth_flow") as mock_flow:
mock_flow.return_value = {"access_token": "ya29.test"}
conn.handle_callback("test-id.apps.googleusercontent.com:test-secret")
mock_flow.assert_called_once()
def test_gdrive_handle_callback_fallback_on_failure(tmp_path: Path) -> None:
from openjarvis.connectors.gdrive import GDriveConnector
from openjarvis.connectors.oauth import load_tokens
creds = str(tmp_path / "gdrive.json")
conn = GDriveConnector(credentials_path=creds)
with patch("openjarvis.connectors.gdrive.run_oauth_flow") as mock_flow:
mock_flow.side_effect = RuntimeError("OAuth failed")
conn.handle_callback("test-id.apps.googleusercontent.com:test-secret")
# Should have saved client_id and client_secret as fallback
tokens = load_tokens(creds)
assert tokens is not None
assert tokens["client_id"] == "test-id.apps.googleusercontent.com"
assert tokens["client_secret"] == "test-secret"
def test_gdrive_handle_callback_raw_token(tmp_path: Path) -> None:
from openjarvis.connectors.gdrive import GDriveConnector
from openjarvis.connectors.oauth import load_tokens
creds = str(tmp_path / "gdrive.json")
conn = GDriveConnector(credentials_path=creds)
conn.handle_callback("some-raw-token-value")
tokens = load_tokens(creds)
assert tokens is not None
assert tokens["token"] == "some-raw-token-value"
def test_gdrive_auth_url_returns_credentials_page_without_client_id(
tmp_path: Path,
) -> None:
from openjarvis.connectors.gdrive import GDriveConnector
creds = str(tmp_path / "gdrive.json")
conn = GDriveConnector(credentials_path=creds)
url = conn.auth_url()
assert url == "https://console.cloud.google.com/apis/credentials"
def test_gdrive_auth_url_returns_consent_url_with_client_id(
tmp_path: Path,
) -> None:
from openjarvis.connectors.gdrive import GDriveConnector
from openjarvis.connectors.oauth import save_tokens
creds = str(tmp_path / "gdrive.json")
conn = GDriveConnector(credentials_path=creds)
save_tokens(creds, {"client_id": "test-id.apps.googleusercontent.com"})
url = conn.auth_url()
assert "accounts.google.com" in url
assert "test-id.apps.googleusercontent.com" in url