""" Flask application for handling the Basecamp 3 OAuth 2.0 authorization flow. This application provides endpoints for: 1. Redirecting users to Basecamp for authorization 2. Handling the OAuth callback 3. Using the obtained token to access the Basecamp API 4. Providing a secure token endpoint for the MCP server """ import os import sys import json import secrets import logging from flask import Flask, request, redirect, url_for, session, render_template_string, jsonify from dotenv import load_dotenv from basecamp_oauth import BasecampOAuth from basecamp_client import BasecampClient from search_utils import BasecampSearch import token_storage # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("oauth_app.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Load environment variables load_dotenv() # Check for required environment variables required_vars = ['BASECAMP_CLIENT_ID', 'BASECAMP_CLIENT_SECRET', 'BASECAMP_REDIRECT_URI', 'USER_AGENT'] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: logger.error(f"Missing required environment variables: {', '.join(missing_vars)}") logger.error("Please set these variables in your .env file or environment") sys.exit(1) # Create Flask app app = Flask(__name__) app.secret_key = os.getenv('FLASK_SECRET_KEY', secrets.token_hex(16)) # HTML template for displaying results RESULTS_TEMPLATE = """ Basecamp 3 OAuth Demo

{% if message %}

{% endif %} {% if warning %}

{% endif %} {% if content %}

{{ content }}

{% endif %} {% if auth_url %} Log in with Basecamp {% endif %} {% if token_info %}

OAuth Token Information

{{ token_info | tojson(indent=2) }}

{% endif %} {% if show_logout %} Logout {% endif %} {% if show_home %} Home {% endif %}

""" @app.template_filter('tojson') def to_json(value, indent=None): return json.dumps(value, indent=indent) def get_oauth_client(): """Get a configured OAuth client.""" try: client_id = os.getenv('BASECAMP_CLIENT_ID') client_secret = os.getenv('BASECAMP_CLIENT_SECRET') redirect_uri = os.getenv('BASECAMP_REDIRECT_URI') user_agent = os.getenv('USER_AGENT') logger.info("Creating OAuth client with config: %s, %s, %s", client_id, redirect_uri, user_agent) return BasecampOAuth( client_id=client_id, client_secret=client_secret, redirect_uri=redirect_uri, user_agent=user_agent ) except Exception as e: logger.error("Error creating OAuth client: %s", str(e)) raise def ensure_valid_token(): """ Ensure we have a valid, non-expired token. Attempts to refresh if expired. Returns: dict: Valid token data or None if authentication is needed """ token_data = token_storage.get_token() if not token_data or not token_data.get('access_token'): logger.info("No token found") return None # Check if token is expired if token_storage.is_token_expired(): logger.info("Token is expired, attempting to refresh") refresh_token = token_data.get('refresh_token') if not refresh_token: logger.warning("No refresh token available, user needs to re-authenticate") return None try: oauth_client = get_oauth_client() new_token_data = oauth_client.refresh_token(refresh_token) # Store the new token access_token = new_token_data.get('access_token') new_refresh_token = new_token_data.get('refresh_token', refresh_token) # Use old refresh token if new one not provided expires_in = new_token_data.get('expires_in') account_id = token_data.get('account_id') # Keep the existing account_id if access_token: token_storage.store_token( access_token=access_token, refresh_token=new_refresh_token, expires_in=expires_in, account_id=account_id ) logger.info("Token refreshed successfully") return token_storage.get_token() else: logger.error("No access token in refresh response") return None except Exception as e: logger.error("Failed to refresh token: %s", str(e)) return None logger.info("Token is valid") return token_data @app.route('/') def home(): """Home page.""" # Ensure we have a valid token token_data = ensure_valid_token() if token_data and token_data.get('access_token'): # We have a valid token, show token information access_token = token_data['access_token'] # Mask the token for security masked_token = f"{access_token[:10]}...{access_token[-10:]}" if len(access_token) > 20 else "***" token_info = { "access_token": masked_token, "account_id": token_data.get('account_id'), "has_refresh_token": bool(token_data.get('refresh_token')), "expires_at": token_data.get('expires_at'), "updated_at": token_data.get('updated_at') } logger.info("Home page: User is authenticated") return render_template_string( RESULTS_TEMPLATE, title="Basecamp OAuth Status", message="You are authenticated with Basecamp!", token_info=token_info, show_logout=True ) else: # No valid token, show login button try: oauth_client = get_oauth_client() auth_url = oauth_client.get_authorization_url() logger.info("Home page: User not authenticated, showing login button") return render_template_string( RESULTS_TEMPLATE, title="Basecamp OAuth Demo", message="Welcome! Please log in with your Basecamp account to continue.", auth_url=auth_url ) except Exception as e: logger.error("Error getting authorization URL: %s", str(e)) return render_template_string( RESULTS_TEMPLATE, title="Error", message=f"Error setting up OAuth: {str(e)}", ) @app.route('/auth/callback') def auth_callback(): """Handle the OAuth callback from Basecamp.""" logger.info("OAuth callback called with args: %s", request.args) code = request.args.get('code') error = request.args.get('error') if error: logger.error("OAuth callback error: %s", error) return render_template_string( RESULTS_TEMPLATE, title="Authentication Error", message=f"Basecamp returned an error: {error}", show_home=True ) if not code: logger.error("OAuth callback: No code provided") return render_template_string( RESULTS_TEMPLATE, title="Error", message="No authorization code received.", show_home=True ) try: # Exchange the code for an access token oauth_client = get_oauth_client() logger.info("Exchanging code for token") token_data = oauth_client.exchange_code_for_token(code) logger.info(f"Raw token data from Basecamp exchange: {token_data}") # Store the token in our secure storage access_token = token_data.get('access_token') refresh_token = token_data.get('refresh_token') expires_in = token_data.get('expires_in') account_id = os.getenv('BASECAMP_ACCOUNT_ID') if not access_token: logger.error("OAuth exchange: No access token received") return render_template_string( RESULTS_TEMPLATE, title="Authentication Error", message="No access token received from Basecamp.", show_home=True ) # Try to get identity if account_id is not set if not account_id: try: logger.info("Getting user identity to find account_id") identity = oauth_client.get_identity(access_token) logger.info("Identity response: %s", identity) # Find Basecamp 3 account if identity.get('accounts'): for account in identity['accounts']: if account.get('product') == 'bc3': # Basecamp 3 account_id = account['id'] logger.info("Found account_id: %s", account_id) break except Exception as identity_error: logger.error("Error getting identity: %s", str(identity_error)) # Continue with the flow, but log the error logger.info("Storing token with account_id: %s", account_id) stored = token_storage.store_token( access_token=access_token, refresh_token=refresh_token, expires_in=expires_in, account_id=account_id ) if not stored: logger.error("Failed to store token") return render_template_string( RESULTS_TEMPLATE, title="Error", message="Failed to store token. Please try again.", show_home=True ) # Also keep the access token in session for convenience session['access_token'] = access_token if refresh_token: session['refresh_token'] = refresh_token if account_id: session['account_id'] = account_id logger.info("OAuth flow completed successfully") return redirect(url_for('home')) except Exception as e: logger.error("Error in OAuth callback: %s", str(e), exc_info=True) return render_template_string( RESULTS_TEMPLATE, title="Error", message=f"Failed to exchange code for token: {str(e)}", show_home=True ) @app.route('/api/token', methods=['GET']) def get_token_api(): """ Secure API endpoint for the MCP server to get the token. This should only be accessible by the MCP server. """ logger.info("Token API called with headers: %s", request.headers) # In production, implement proper authentication for this endpoint # For now, we'll use a simple API key check api_key = request.headers.get('X-API-Key') if not api_key or api_key != os.getenv('MCP_API_KEY', 'mcp_secret_key'): logger.error("Token API: Invalid API key") return jsonify({ "error": "Unauthorized", "message": "Invalid or missing API key" }), 401 # Use the ensure_valid_token function to get a fresh token token_data = ensure_valid_token() if not token_data or not token_data.get('access_token'): logger.error("Token API: No valid token available") return jsonify({ "error": "Not authenticated", "message": "No valid token available" }), 404 logger.info("Token API: Successfully returned token") return jsonify({ "access_token": token_data['access_token'], "account_id": token_data.get('account_id') }) @app.route('/logout') def logout(): """Clear the session and token storage.""" logger.info("Logout called") session.clear() token_storage.clear_tokens() return redirect(url_for('home')) @app.route('/token/info') def token_info(): """Display information about the stored token.""" logger.info("Token info called") token_data = token_storage.get_token() if not token_data: logger.info("Token info: No token stored") return render_template_string( RESULTS_TEMPLATE, title="Token Information", message="No token stored.", show_home=True ) # Check if token is expired is_expired = token_storage.is_token_expired() # Mask the tokens for security access_token = token_data.get('access_token', '') refresh_token = token_data.get('refresh_token', '') masked_access = f"{access_token[:10]}...{access_token[-10:]}" if len(access_token) > 20 else "***" masked_refresh = f"{refresh_token[:10]}...{refresh_token[-10:]}" if refresh_token and len(refresh_token) > 20 else "***" if refresh_token else None display_info = { "access_token": masked_access, "has_refresh_token": bool(refresh_token), "account_id": token_data.get('account_id'), "expires_at": token_data.get('expires_at'), "updated_at": token_data.get('updated_at'), "is_expired": is_expired } warning_message = None if is_expired: warning_message = "Warning: Your token is expired! Visit the home page to automatically refresh it, or logout and log back in." logger.info("Token info: Returned token info") return render_template_string( RESULTS_TEMPLATE, title="Token Information", content=json.dumps(display_info, indent=2), warning=warning_message, show_home=True ) @app.route('/health') def health_check(): """Health check endpoint.""" logger.info("Health check called") return jsonify({ "status": "ok", "service": "basecamp-oauth-app" }) if __name__ == '__main__': try: logger.info("Starting OAuth app on port %s", os.environ.get('PORT', 8000)) # Run the Flask app port = int(os.environ.get('PORT', 8000)) # Disable debug and auto-reloader when running in production or background is_debug = os.environ.get('FLASK_DEBUG', 'False').lower() == 'true' logger.info("Running in %s mode", "debug" if is_debug else "production") app.run(host='0.0.0.0', port=port, debug=is_debug, use_reloader=is_debug) except Exception as e: logger.error("Fatal error: %s", str(e), exc_info=True) sys.exit(1)

{{ title }}

OAuth Token Information