""" 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 from config_paths import get_log_directory, get_env_file_path, ensure_directories_exist ensure_directories_exist() DOTENV_PATH = get_env_file_path() if DOTENV_PATH.exists(): load_dotenv(DOTENV_PATH) LOG_FILE_PATH = get_log_directory() / "oauth_app.log" logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(LOG_FILE_PATH), logging.StreamHandler(), ], ) logger = logging.getLogger(__name__) # 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