tessl/pypi-jupyterhub
A multi-user server for Jupyter notebooks that provides authentication, spawning, and proxying for multiple users simultaneously
—
Pending
services-oauth.mddocs/
Services and OAuth Integration
JupyterHub provides comprehensive support for integrating external services through managed and unmanaged services, along with OAuth 2.0 provider capabilities for third-party application integration. This system enables secure authentication and authorization for external applications and services.
Capabilities
Service Management
Core functionality for registering and managing external services with JupyterHub.
class Service:
"""
JupyterHub service for external application integration.
Services can be managed (started/stopped by JupyterHub) or
unmanaged (external processes that authenticate with JupyterHub).
"""
# Service configuration
name: str # Unique service name
url: str # Service URL (for unmanaged services)
prefix: str # URL prefix for routing requests
command: List[str] # Command to start managed service
environment: Dict[str, str] # Environment variables
# Service properties
admin: bool # Whether service has admin privileges
managed: bool # Whether JupyterHub manages the service process
api_token: str # API token for service authentication
oauth_client_id: str # OAuth client ID (if using OAuth)
def __init__(self, **kwargs):
"""
Initialize service configuration.
Args:
**kwargs: Service configuration parameters
"""
def start(self):
"""Start a managed service process"""
def stop(self):
"""Stop a managed service process"""
@property
def pid(self) -> int:
"""Process ID of managed service"""
@property
def proc(self) -> subprocess.Popen:
"""Process object for managed service"""
# Service configuration in jupyterhub_config.py
c.JupyterHub.services = [
{
'name': 'my-service',
'url': 'http://localhost:8001',
'api_token': 'secret-token',
'admin': True
}
]Hub Authentication for Services
Authentication utilities for services to authenticate with JupyterHub.
class HubAuth:
"""
Authentication helper for JupyterHub services.
Provides methods for services to authenticate requests
and interact with the JupyterHub API.
"""
def __init__(self,
api_token: str = None,
api_url: str = None,
cache_max_age: int = 300):
"""
Initialize Hub authentication.
Args:
api_token: Service API token
api_url: JupyterHub API URL
cache_max_age: Cache duration for user info
"""
async def user_for_token(self, token: str, sync: bool = True) -> Dict[str, Any]:
"""
Get user information for an API token.
Args:
token: API token to validate
sync: Whether to sync with database
Returns:
User information dictionary or None if invalid
"""
async def user_for_cookie(self, cookie_name: str, cookie_value: str, use_cache: bool = True) -> Dict[str, Any]:
"""
Get user information for a login cookie.
Args:
cookie_name: Name of the cookie
cookie_value: Cookie value
use_cache: Whether to use cached results
Returns:
User information dictionary or None if invalid
"""
async def api_request(self, method: str, url: str, **kwargs) -> requests.Response:
"""
Make authenticated request to JupyterHub API.
Args:
method: HTTP method (GET, POST, etc.)
url: API endpoint URL (relative to api_url)
**kwargs: Additional request parameters
Returns:
HTTP response object
"""
# Example service using HubAuth
from jupyterhub.services.auth import HubAuth
auth = HubAuth(
api_token=os.environ['JUPYTERHUB_API_TOKEN'],
api_url=os.environ['JUPYTERHUB_API_URL']
)
@app.route('/dashboard')
async def dashboard():
"""Protected service endpoint"""
cookie = request.cookies.get('jupyterhub-hub-login')
user = await auth.user_for_cookie('jupyterhub-hub-login', cookie)
if not user:
return redirect('/hub/login')
return render_template('dashboard.html', user=user)OAuth Provider System
JupyterHub's OAuth 2.0 provider implementation for third-party application integration.
class OAuthProvider:
"""
OAuth 2.0 authorization server implementation.
Enables third-party applications to obtain access tokens
for JupyterHub API access.
"""
# OAuth endpoints (handled by JupyterHub)
# GET /hub/api/oauth2/authorize - Authorization endpoint
# POST /hub/api/oauth2/token - Token endpoint
# GET /hub/api/oauth2/userinfo - User info endpoint
def generate_authorization_code(self, client_id: str, user: User, scopes: List[str]) -> str:
"""
Generate OAuth authorization code.
Args:
client_id: OAuth client ID
user: Authorizing user
scopes: Requested scopes
Returns:
Authorization code string
"""
def exchange_code_for_token(self, code: str, client_id: str, client_secret: str) -> Dict[str, Any]:
"""
Exchange authorization code for access token.
Args:
code: Authorization code
client_id: OAuth client ID
client_secret: OAuth client secret
Returns:
Token response with access_token, token_type, scope
"""
def get_user_info(self, access_token: str) -> Dict[str, Any]:
"""
Get user information for access token.
Args:
access_token: OAuth access token
Returns:
User information (subject, name, groups, etc.)
"""
# OAuth client registration
class OAuthClient(Base):
"""OAuth client application registration"""
id: str # Client ID (primary key)
identifier: str # Human-readable identifier
description: str # Client description
secret: str # Client secret (hashed)
redirect_uri: str # Authorized redirect URI
allowed_scopes: List[str] # Scopes client can request
def check_secret(self, secret: str) -> bool:
"""Verify client secret"""
def check_redirect_uri(self, uri: str) -> bool:
"""Verify redirect URI is authorized"""Usage Examples
Basic Service Configuration
# jupyterhub_config.py
# Unmanaged service (external process)
c.JupyterHub.services = [
{
'name': 'announcement-service',
'url': 'http://localhost:8001',
'api_token': 'your-secret-token-here',
'admin': False,
'oauth_redirect_uri': 'http://localhost:8001/oauth-callback'
}
]
# Managed service (started by JupyterHub)
c.JupyterHub.services = [
{
'name': 'monitoring-service',
'managed': True,
'command': ['python', '/path/to/monitoring_service.py'],
'environment': {
'JUPYTERHUB_SERVICE_NAME': 'monitoring-service',
'JUPYTERHUB_SERVICE_URL': 'http://127.0.0.1:8002'
},
'url': 'http://127.0.0.1:8002',
'api_token': 'monitoring-token'
}
]Service Implementation Example
# announcement_service.py
import os
from flask import Flask, request, redirect, render_template
from jupyterhub.services.auth import HubAuth
app = Flask(__name__)
# Initialize Hub authentication
auth = HubAuth(
api_token=os.environ['JUPYTERHUB_API_TOKEN'],
api_url=os.environ['JUPYTERHUB_API_URL']
)
@app.route('/')
async def index():
"""Main service page with user authentication"""
# Get user from cookie
cookie = request.cookies.get('jupyterhub-hub-login')
if not cookie:
return redirect('/hub/login?next=' + request.url)
user = await auth.user_for_cookie('jupyterhub-hub-login', cookie)
if not user:
return redirect('/hub/login?next=' + request.url)
# Get announcements for user
announcements = get_announcements_for_user(user)
return render_template('announcements.html',
user=user,
announcements=announcements)
@app.route('/api/announcements')
async def api_announcements():
"""API endpoint for announcements"""
# Authenticate via API token
token = request.headers.get('Authorization', '').replace('Bearer ', '')
user = await auth.user_for_token(token)
if not user:
return {'error': 'Unauthorized'}, 401
return {'announcements': get_announcements_for_user(user)}
def get_announcements_for_user(user):
"""Get announcements relevant to user"""
# Implementation depends on your announcement system
return []
if __name__ == '__main__':
app.run(port=8001)OAuth Client Registration
# Register OAuth client application
from jupyterhub.orm import OAuthClient
client = OAuthClient(
id='my-external-app',
identifier='My External Application',
description='Third-party app that integrates with JupyterHub',
redirect_uri='https://myapp.example.com/oauth/callback',
allowed_scopes=['read:users', 'read:servers', 'identify']
)
# Set client secret (will be hashed)
client.secret = 'your-client-secret-here'
# Add to database
db.add(client)
db.commit()OAuth Flow Implementation
# External application OAuth integration
import requests
from urllib.parse import urlencode
class JupyterHubOAuth:
"""OAuth client for JupyterHub integration"""
def __init__(self, client_id, client_secret, hub_url):
self.client_id = client_id
self.client_secret = client_secret
self.hub_url = hub_url
def get_authorization_url(self, redirect_uri, scopes, state=None):
"""
Generate OAuth authorization URL.
Args:
redirect_uri: Where to redirect after authorization
scopes: List of requested scopes
state: CSRF protection state parameter
Returns:
Authorization URL string
"""
params = {
'client_id': self.client_id,
'redirect_uri': redirect_uri,
'scope': ' '.join(scopes),
'response_type': 'code'
}
if state:
params['state'] = state
return f"{self.hub_url}/hub/api/oauth2/authorize?{urlencode(params)}"
async def exchange_code_for_token(self, code, redirect_uri):
"""
Exchange authorization code for access token.
Args:
code: Authorization code from callback
redirect_uri: Original redirect URI
Returns:
Token response dictionary
"""
token_url = f"{self.hub_url}/hub/api/oauth2/token"
data = {
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': redirect_uri,
'client_id': self.client_id,
'client_secret': self.client_secret
}
response = requests.post(token_url, data=data)
return response.json()
async def get_user_info(self, access_token):
"""
Get user information with access token.
Args:
access_token: OAuth access token
Returns:
User information dictionary
"""
headers = {'Authorization': f'Bearer {access_token}'}
response = requests.get(
f"{self.hub_url}/hub/api/oauth2/userinfo",
headers=headers
)
return response.json()
# Usage in web application
oauth = JupyterHubOAuth(
client_id='my-external-app',
client_secret='your-client-secret',
hub_url='https://hub.example.com'
)
# Redirect user to authorization
auth_url = oauth.get_authorization_url(
redirect_uri='https://myapp.example.com/oauth/callback',
scopes=['read:users', 'identify'],
state='csrf-protection-token'
)
# Handle callback
@app.route('/oauth/callback')
async def oauth_callback():
code = request.args.get('code')
state = request.args.get('state')
# Verify state for CSRF protection
if state != session.get('oauth_state'):
return 'Invalid state', 400
# Exchange code for token
token_response = await oauth.exchange_code_for_token(
code=code,
redirect_uri='https://myapp.example.com/oauth/callback'
)
# Get user information
user_info = await oauth.get_user_info(token_response['access_token'])
# Store token and user info in session
session['access_token'] = token_response['access_token']
session['user'] = user_info
return redirect('/dashboard')Service with Role-Based Access
# Service with RBAC integration
from jupyterhub.services.auth import HubAuth
from jupyterhub.scopes import check_scopes
class RBACService:
"""Service with role-based access control"""
def __init__(self):
self.auth = HubAuth()
async def check_permission(self, token, required_scopes):
"""
Check if token has required permissions.
Args:
token: API token or cookie
required_scopes: List of required scopes
Returns:
User info if authorized, None otherwise
"""
user = await self.auth.user_for_token(token)
if not user:
return None
user_scopes = user.get('scopes', [])
if check_scopes(required_scopes, user_scopes):
return user
return None
# Usage in service endpoints
service = RBACService()
@app.route('/admin/users')
async def admin_users():
"""Admin-only endpoint"""
token = request.headers.get('Authorization', '').replace('Bearer ', '')
user = await service.check_permission(token, ['admin:users'])
if not user:
return {'error': 'Insufficient permissions'}, 403
# Admin functionality here
return {'users': []}Advanced Integration Patterns
Multi-Service Coordination
# Service registry for coordinating multiple services
class ServiceRegistry:
"""Registry for coordinating multiple JupyterHub services"""
def __init__(self, hub_auth):
self.auth = hub_auth
self.services = {}
async def register_service(self, name, url, capabilities):
"""Register a service with capabilities"""
# Verify service authentication
service_info = await self.auth.api_request('GET', f'/services/{name}')
if service_info.status_code == 200:
self.services[name] = {
'url': url,
'capabilities': capabilities,
'info': service_info.json()
}
async def discover_service(self, capability):
"""Find services with specific capability"""
return [
service for service in self.services.values()
if capability in service['capabilities']
]
# Service mesh configuration
c.JupyterHub.services = [
{
'name': 'service-registry',
'managed': True,
'command': ['python', '/path/to/service_registry.py'],
'url': 'http://127.0.0.1:8003',
'admin': True
},
{
'name': 'data-service',
'url': 'http://localhost:8004',
'capabilities': ['data-processing', 'file-storage']
},
{
'name': 'compute-service',
'url': 'http://localhost:8005',
'capabilities': ['job-execution', 'resource-management']
}
]Event-Driven Service Integration
# Service with event handling
import asyncio
from jupyterhub.services.auth import HubAuth
class EventDrivenService:
"""Service that responds to JupyterHub events"""
def __init__(self):
self.auth = HubAuth()
self.event_queue = asyncio.Queue()
async def poll_events(self):
"""Poll JupyterHub for events"""
while True:
try:
# Check for user activity updates
response = await self.auth.api_request('GET', '/users')
users = response.json()
for user in users:
if self.should_process_user(user):
await self.event_queue.put({
'type': 'user_activity',
'user': user
})
await asyncio.sleep(30) # Poll every 30 seconds
except Exception as e:
print(f"Error polling events: {e}")
await asyncio.sleep(60)
async def process_events(self):
"""Process events from queue"""
while True:
event = await self.event_queue.get()
await self.handle_event(event)
async def handle_event(self, event):
"""Handle specific event types"""
if event['type'] == 'user_activity':
await self.update_user_metrics(event['user'])
def should_process_user(self, user):
"""Determine if user event should be processed"""
# Your event filtering logic
return True
async def update_user_metrics(self, user):
"""Update metrics for user activity"""
# Your metrics update logic
passInstall with Tessl CLI
npx tessl i tessl/pypi-jupyterhub