17249640673

Committed 26 Aug 2025 08:27PM UTC coverage: 92.699% (+0.03%) from 92.672%

Build # 17249640673

Build Type

Pull #7

github

Committed by

web-flow

Commit Message

Merge 4144c5b04 into 653847af2

Pull Request Pull Request #7: Refactor CLI

Run Details

209 of 259 new or added lines in 5 files covered. (80.69%)

3 existing lines in 1 file now uncovered.

838 of 904 relevant lines covered (92.7%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.12

/src/monzoh/cli/auth_flow.py

"""Main authentication flow orchestration."""

import secrets
import urllib.parse
import webbrowser

from rich.console import Console
from rich.panel import Panel
from rich.text import Text

from ..auth import MonzoOAuth
from ..main import MonzoClient
from .credentials import (
    get_credentials_interactively,
    load_env_credentials,
    save_credentials_to_env,
)
from .oauth_server import start_callback_server
from .token_cache import (
    clear_token_cache,
    load_token_from_cache,
    save_token_to_cache,
    try_refresh_token,
)


def authenticate() -> str | None:
    """Main authentication flow."""
    console = Console()

    # Show welcome banner
    console.print()
    console.print(
        Panel(
            Text(
                "🏦 Monzo API Authentication Tool",
                style="bold magenta",
                justify="center",
            ),
            style="magenta",
        )
    )

    try:
        # Check for cached token first (including expired ones)
        cached_token = load_token_from_cache(include_expired=True)
        if cached_token:
            console.print("🔍 Found cached access token")

            # Test if token is still valid
            console.print("🧪 Testing cached token...")
            try:
                with MonzoClient(cached_token["access_token"]) as client:
                    whoami = client.whoami()
                    console.print(
                        f"✅ [green]Using cached token for: {whoami.user_id}[/green]"
                    )
                    access_token: str = cached_token["access_token"]
                    return access_token
            except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):
                console.print("❌ Error during authentication: Token is invalid")

                # Try to refresh the token
                existing_creds = load_env_credentials()
                if existing_creds.get("client_id") and existing_creds.get(
                    "client_secret"
                ):
                    client_id = existing_creds["client_id"]
                    client_secret = existing_creds["client_secret"]
                    redirect_uri = existing_creds.get(
                        "redirect_uri", "http://localhost:8080/callback"
                    )
                    assert client_id is not None
                    assert client_secret is not None
                    assert redirect_uri is not None  # get() with default never None

                    oauth = MonzoOAuth(
                        client_id=client_id,
                        client_secret=client_secret,
                        redirect_uri=redirect_uri,
                    )

                    refreshed_token = try_refresh_token(cached_token, oauth, console)
                    if refreshed_token:
                        return refreshed_token

                # Clear invalid cache
                clear_token_cache()
                console.print("🗑️  Cleared invalid token cache")

        # Load existing credentials
        existing_creds = load_env_credentials()

        # Get missing credentials interactively
        creds = get_credentials_interactively(console, existing_creds)

        # Offer to save credentials
        if not all(existing_creds.get(k) for k in ["client_id", "client_secret"]):
            save_credentials_to_env(creds, console)

        # Extract port from redirect URI
        parsed_uri = urllib.parse.urlparse(creds["redirect_uri"])
        port = parsed_uri.port or 8080

        # Start callback server
        console.print(f"\n🚀 Starting callback server on port [cyan]{port}[/cyan]...")
        server = start_callback_server(port)

        # Create OAuth client
        oauth = MonzoOAuth(
            client_id=creds["client_id"],
            client_secret=creds["client_secret"],
            redirect_uri=creds["redirect_uri"],
        )

        # Generate state for CSRF protection
        state = secrets.token_urlsafe(32)
        auth_url = oauth.get_authorization_url(state=state)

        console.print("\n📋 Next steps:")
        console.print(
            "1. A browser window will open with the Monzo authorization page "
            "(if not, you'll need to do it manually)"
        )
        console.print("2. Log in to your Monzo account and authorize the application")
        console.print("3. You'll be redirected back automatically")

        # Open browser automatically
        webbrowser.open(auth_url)
        console.print(
            f"\nIf your browser does not open automatically, use the following "
            f"link to authenticate:\n[blue]{auth_url}[/blue]"
        )

        console.print("\n⏳ Waiting for authorization... (Press Ctrl+C to cancel)")

        # Wait for callback
        callback_timeout = 300  # 5 minutes
        if server.callback_received.wait(timeout=callback_timeout):
            server.shutdown()

            if server.error:
                console.print(f"\n❌ [red]Authorization failed: {server.error}[/red]")
                return None

            if not server.auth_code:
                console.print("\n❌ [red]No authorization code received[/red]")
                return None

            if server.state != state:
                console.print(
                    "\n❌ [red]Invalid state parameter - possible CSRF attack[/red]"
                )
                return None

            console.print("\n✅ [green]Authorization code received![/green]")
            console.print("🔄 Exchanging code for access token...")

            # Exchange code for token
            with oauth:
                token = oauth.exchange_code_for_token(server.auth_code)

            console.print("🎉 [bold green]Authentication successful![/bold green]")
            console.print(f"Access Token: [green]{token.access_token[:20]}...[/green]")

            # Save token to cache
            save_token_to_cache(token, console)

            # Test the token
            console.print("\n🧪 Testing API access...")
            with MonzoClient(token.access_token) as client:
                whoami = client.whoami()
                console.print(f"✅ Connected as: [cyan]{whoami.user_id}[/cyan]")

            return token.access_token

        else:
            server.shutdown()
            console.print(
                f"\n⏰ [yellow]Timeout after {callback_timeout} seconds[/yellow]"
            )
            return None

    except KeyboardInterrupt:
        console.print("\n\n👋 [yellow]Authentication cancelled by user[/yellow]")
        return None
    except Exception as e:
        console.print(f"\n❌ [red]Error during authentication: {e}[/red]")
        return None

1	"""Main authentication flow orchestration."""
2
3	import secrets	1✔
4	import urllib.parse	1✔
5	import webbrowser	1✔
6
7	from rich.console import Console	1✔
8	from rich.panel import Panel	1✔
9	from rich.text import Text	1✔
10
11	from ..auth import MonzoOAuth	1✔
12	from ..main import MonzoClient	1✔
13	from .credentials import (	1✔
14	get_credentials_interactively,
15	load_env_credentials,
16	save_credentials_to_env,
17	)
18	from .oauth_server import start_callback_server	1✔
19	from .token_cache import (	1✔
20	clear_token_cache,
21	load_token_from_cache,
22	save_token_to_cache,
23	try_refresh_token,
24	)
25
26
27	def authenticate() -> str \| None:	1✔
28	"""Main authentication flow."""
29	console = Console()	1✔
30
31	# Show welcome banner
32	console.print()	1✔
33	console.print(	1✔
34	Panel(
35	Text(
36	"🏦 Monzo API Authentication Tool",
37	style="bold magenta",
38	justify="center",
39	),
40	style="magenta",
41	)
42	)
43
44	try:	1✔
45	# Check for cached token first (including expired ones)
46	cached_token = load_token_from_cache(include_expired=True)	1✔
47	if cached_token:	1✔
48	console.print("🔍 Found cached access token")	1✔
49
50	# Test if token is still valid
51	console.print("🧪 Testing cached token...")	1✔
52	try:	1✔
53	with MonzoClient(cached_token["access_token"]) as client:	1✔
54	whoami = client.whoami()	1✔
55	console.print(	1✔
56	f"✅ [green]Using cached token for: {whoami.user_id}[/green]"
57	)
58	access_token: str = cached_token["access_token"]	1✔
59	return access_token	1✔
NEW 60	except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):	×
NEW 61	console.print("❌ Error during authentication: Token is invalid")	×
62
63	# Try to refresh the token
NEW 64	existing_creds = load_env_credentials()	×
NEW 65	if existing_creds.get("client_id") and existing_creds.get(	×
66	"client_secret"
67	):
NEW 68	client_id = existing_creds["client_id"]	×
NEW 69	client_secret = existing_creds["client_secret"]	×
NEW 70	redirect_uri = existing_creds.get(	×
71	"redirect_uri", "http://localhost:8080/callback"
72	)
NEW 73	assert client_id is not None	×
NEW 74	assert client_secret is not None	×
NEW 75	assert redirect_uri is not None # get() with default never None	×
76
NEW 77	oauth = MonzoOAuth(	×
78	client_id=client_id,
79	client_secret=client_secret,
80	redirect_uri=redirect_uri,
81	)
82
NEW 83	refreshed_token = try_refresh_token(cached_token, oauth, console)	×
NEW 84	if refreshed_token:	×
NEW 85	return refreshed_token	×
86
87	# Clear invalid cache
NEW 88	clear_token_cache()	×
NEW 89	console.print("🗑️ Cleared invalid token cache")	×
90
91	# Load existing credentials
92	existing_creds = load_env_credentials()	1✔
93
94	# Get missing credentials interactively
95	creds = get_credentials_interactively(console, existing_creds)	1✔
96
97	# Offer to save credentials
98	if not all(existing_creds.get(k) for k in ["client_id", "client_secret"]):	1✔
99	save_credentials_to_env(creds, console)	1✔
100
101	# Extract port from redirect URI
102	parsed_uri = urllib.parse.urlparse(creds["redirect_uri"])	1✔
103	port = parsed_uri.port or 8080	1✔
104
105	# Start callback server
106	console.print(f"\n🚀 Starting callback server on port [cyan]{port}[/cyan]...")	1✔
107	server = start_callback_server(port)	1✔
108
109	# Create OAuth client
110	oauth = MonzoOAuth(	1✔
111	client_id=creds["client_id"],
112	client_secret=creds["client_secret"],
113	redirect_uri=creds["redirect_uri"],
114	)
115
116	# Generate state for CSRF protection
117	state = secrets.token_urlsafe(32)	1✔
118	auth_url = oauth.get_authorization_url(state=state)	1✔
119
120	console.print("\n📋 Next steps:")	1✔
121	console.print(	1✔
122	"1. A browser window will open with the Monzo authorization page "
123	"(if not, you'll need to do it manually)"
124	)
125	console.print("2. Log in to your Monzo account and authorize the application")	1✔
126	console.print("3. You'll be redirected back automatically")	1✔
127
128	# Open browser automatically
129	webbrowser.open(auth_url)	1✔
130	console.print(	1✔
131	f"\nIf your browser does not open automatically, use the following "
132	f"link to authenticate:\n[blue]{auth_url}[/blue]"
133	)
134
135	console.print("\n⏳ Waiting for authorization... (Press Ctrl+C to cancel)")	1✔
136
137	# Wait for callback
138	callback_timeout = 300 # 5 minutes	1✔
139	if server.callback_received.wait(timeout=callback_timeout):	1✔
140	server.shutdown()	1✔
141
142	if server.error:	1✔
NEW 143	console.print(f"\n❌ [red]Authorization failed: {server.error}[/red]")	×
NEW 144	return None	×
145
146	if not server.auth_code:	1✔
NEW 147	console.print("\n❌ [red]No authorization code received[/red]")	×
NEW 148	return None	×
149
150	if server.state != state:	1✔
NEW 151	console.print(	×
152	"\n❌ [red]Invalid state parameter - possible CSRF attack[/red]"
153	)
NEW 154	return None	×
155
156	console.print("\n✅ [green]Authorization code received![/green]")	1✔
157	console.print("🔄 Exchanging code for access token...")	1✔
158
159	# Exchange code for token
160	with oauth:	1✔
161	token = oauth.exchange_code_for_token(server.auth_code)	1✔
162
163	console.print("🎉 [bold green]Authentication successful![/bold green]")	1✔
164	console.print(f"Access Token: [green]{token.access_token[:20]}...[/green]")	1✔
165
166	# Save token to cache
167	save_token_to_cache(token, console)	1✔
168
169	# Test the token
170	console.print("\n🧪 Testing API access...")	1✔
171	with MonzoClient(token.access_token) as client:	1✔
172	whoami = client.whoami()	1✔
173	console.print(f"✅ Connected as: [cyan]{whoami.user_id}[/cyan]")	1✔
174
175	return token.access_token	1✔
176
177	else:
178	server.shutdown()	1✔
179	console.print(	1✔
180	f"\n⏰ [yellow]Timeout after {callback_timeout} seconds[/yellow]"
181	)
182	return None	1✔
183
184	except KeyboardInterrupt:	1✔
185	console.print("\n\n👋 [yellow]Authentication cancelled by user[/yellow]")	1✔
186	return None	1✔
NEW 187	except Exception as e:	×
NEW 188	console.print(f"\n❌ [red]Error during authentication: {e}[/red]")	×
NEW 189	return None	×

samdobson / monzoh / 17249640673

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous