Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,14 +28,17 @@ A fast CLI for Microsoft **Tasks**, **Calendar**, **Email**, **Contacts**, and *
# Install (Python 3.10+)
pip install git+https://github.com/signalclaude/outpost.git

# Connect your Microsoft account
# Connect your Microsoft account (opens your browser to sign in)
outpost setup

# You're ready
outpost task list
outpost cal today
outpost mail list --unread
outpost teams chats

# Re-authenticate any time (e.g. after a token expires)
outpost auth login
```

## What Can It Do?
Expand All @@ -57,7 +60,7 @@ Every command supports `--output json` for scripting and AI agents.
See the **[Wiki](https://github.com/signalclaude/outpost/wiki)** for full documentation:

- [Installation](https://github.com/signalclaude/outpost/wiki/Installation) — pip, wheel, from source
- [Setup & Authentication](https://github.com/signalclaude/outpost/wiki/Setup-&-Authentication) — device code flow, token storage
- [Setup & Authentication](https://github.com/signalclaude/outpost/wiki/Setup-&-Authentication) — browser sign-in, token storage
- [Commands](https://github.com/signalclaude/outpost/wiki/Commands--Tasks) — Tasks, Calendar, Email, Contacts, Teams
- [MCP Server](https://github.com/signalclaude/outpost/wiki/MCP-Server--Local-Setup) — Claude Desktop integration (39 tools)
- [Remote Access](https://github.com/signalclaude/outpost/wiki/MCP-Server--Remote-Access) — SSE/streamable-http for mobile/LAN
Expand Down
77 changes: 69 additions & 8 deletions src/outpost/auth.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Authentication module for outpost CLI (MSAL device code flow)."""
"""Authentication module for outpost CLI (MSAL interactive + device code flow)."""

import sys

Expand Down Expand Up @@ -66,9 +66,52 @@ def _get_msal_app() -> msal.PublicClientApplication | None:
return _app_instances[profile]


def login_browser(scopes: list[str] | None = None) -> bool:
"""Interactive login via the system browser (auth code + PKCE).

Opens the default browser and captures the redirect on a loopback port.
Unlike the device code flow, this is NOT blocked by Conditional Access
policies that restrict device-code authentication (AADSTS error: "an
authentication flow that is restricted by your admin"). Prefer this flow.

Requires the Azure app registration to list ``http://localhost`` as a
redirect URI under "Mobile and desktop applications" (public client).

Args:
scopes: OAuth scopes to request. Defaults to get_active_scopes().

Returns True on success, False on failure.
"""
app = _get_msal_app()
if app is None:
stderr.print(
"[yellow]Azure AD app not configured yet.[/yellow]\n"
"Set DEFAULT_CLIENT_ID in outpost/config.py after registering your Azure AD app."
)
return False

if scopes is None:
scopes = get_active_scopes()

stderr.print("\n[bold]Opening your browser to sign in...[/bold]")
stderr.print("[dim]If it doesn't open, check for a browser window or tab.[/dim]")

result = app.acquire_token_interactive(scopes=scopes)
if "access_token" in result:
stderr.print("[green]Successfully logged in![/green]")
return True

stderr.print(f"[red]Login failed:[/red] {result.get('error_description', 'unknown error')}")
return False


def login_interactive(scopes: list[str] | None = None) -> bool:
"""Run the device code flow for interactive login.

Note: many tenants block this flow via Conditional Access. Prefer
``login_browser`` unless a headless/no-browser environment requires
device code.

Args:
scopes: OAuth scopes to request. Defaults to get_active_scopes().

Expand Down Expand Up @@ -128,15 +171,33 @@ def require_token() -> str:


def get_auth_status() -> dict:
"""Return current auth status as a dict."""
"""Return current auth status as a dict.

Validates the cached token rather than just checking for a cached account.
A cached account whose refresh token has expired (90-day inactivity) or
no longer covers the active scopes is reported as not logged in, with a
message telling the user to re-authenticate.
"""
app = _get_msal_app()
if app is None:
return {"logged_in": False, "username": None, "message": "Azure AD app not configured"}

accounts = app.get_accounts()
if accounts:
return {
"logged_in": True,
"username": accounts[0].get("username", "unknown"),
}
return {"logged_in": False, "username": None}
if not accounts:
return {"logged_in": False, "username": None, "message": "No account signed in"}

username = accounts[0].get("username", "unknown")
result = app.acquire_token_silent_with_error(get_active_scopes(), account=accounts[0])
if result and "access_token" in result:
return {"logged_in": True, "username": username}

# Account is cached but the token can't be silently obtained — expired
# refresh token or scope change. Surface the underlying reason.
detail = result.get("error_description", "session expired") if result else "session expired"
first_line = detail.splitlines()[0] if detail else "session expired"
return {
"logged_in": False,
"username": username,
"message": f"Session expired for {username} — run 'outpost auth login' to re-authenticate.",
"detail": first_line,
}
34 changes: 30 additions & 4 deletions src/outpost/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -991,7 +991,7 @@ def wrapper(*args, **kwargs):
@handle_errors
def setup():
"""Interactive first-time setup wizard."""
from outpost.auth import login_interactive
from outpost.auth import login_browser
from outpost.config import load_config, save_config, get_active_scopes, get_workspace_dir

stderr.print("[bold]Welcome to Outpost![/bold]")
Expand Down Expand Up @@ -1034,7 +1034,7 @@ def setup():
get_workspace_dir()

scopes = get_active_scopes(config)
success = login_interactive(scopes=scopes)
success = login_browser(scopes=scopes)

if success:
if "teams" in enabled:
Expand All @@ -1043,6 +1043,30 @@ def setup():
stderr.print("[dim]Teams access:[/dim] disabled")


@auth_app.command("login")
@handle_errors
def auth_login(
device: bool = typer.Option(
False, "--device", help="Use device-code flow instead of the browser (headless/no-browser only)"
),
):
"""Sign in (or re-authenticate) without running the full setup wizard.

Defaults to interactive browser login, which is not blocked by Conditional
Access policies that restrict device-code authentication.
"""
from outpost.auth import login_browser, login_interactive
from outpost.config import get_active_scopes

scopes = get_active_scopes()
if device:
success = login_interactive(scopes=scopes)
else:
success = login_browser(scopes=scopes)
if not success:
raise typer.Exit(1)


@auth_app.command("status")
@handle_errors
def auth_status(
Expand All @@ -1060,8 +1084,10 @@ def auth_status(
else:
if status["logged_in"]:
stderr.print(f"[green]Logged in[/green] as [bold]{status['username']}[/bold]")
elif status.get("message"):
stderr.print(f"[yellow]{status['message']}[/yellow]")
else:
stderr.print("[yellow]Not logged in.[/yellow] Run [bold]outpost setup[/bold] to connect.")
stderr.print("[yellow]Not logged in.[/yellow] Run [bold]outpost auth login[/bold] to connect.")


# ── MCP commands ────────────────────────────────────────────────────────────
Expand Down Expand Up @@ -1091,7 +1117,7 @@ def mcp_serve(
stderr.print(f"API key required — run [bold]outpost mcp key[/bold] to view")
server.run(transport=transport, host=host, port=port)
else:
mcp.run()
mcp.run(show_banner=False)


@mcp_app.command("key")
Expand Down
10 changes: 10 additions & 0 deletions src/outpost/mcp_server.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,15 @@
"""MCP server exposing Outpost operations as tools for Claude Desktop."""

import os

# Disable FastMCP's startup network update-check and banner BEFORE importing
# fastmcp — its global settings are read at import time. The update-check makes
# a network request during stdio startup, which can stall the transport past
# Claude Desktop's init timeout on a slow/offline network and surface as
# "can't connect to MCP server". Both are opt-outable via env; we force them off.
os.environ.setdefault("FASTMCP_CHECK_FOR_UPDATES", "off")
os.environ.setdefault("FASTMCP_SHOW_SERVER_BANNER", "false")

import base64
from datetime import datetime, timedelta
from typing import Optional
Expand Down
17 changes: 16 additions & 1 deletion tests/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,29 @@ def test_no_accounts(self):
status = get_auth_status()
assert status["logged_in"] is False

def test_with_account(self):
def test_with_account_valid_token(self):
mock_app = MagicMock()
mock_app.get_accounts.return_value = [{"username": "user@example.com"}]
mock_app.acquire_token_silent_with_error.return_value = {"access_token": "tok"}
with patch("outpost.auth._get_msal_app", return_value=mock_app):
status = get_auth_status()
assert status["logged_in"] is True
assert status["username"] == "user@example.com"

def test_with_account_expired_token(self):
"""Cached account but expired refresh token -> not logged in, with message."""
mock_app = MagicMock()
mock_app.get_accounts.return_value = [{"username": "user@example.com"}]
mock_app.acquire_token_silent_with_error.return_value = {
"error": "invalid_grant",
"error_description": "AADSTS700082: The refresh token has expired.\nTrace ID: x",
}
with patch("outpost.auth._get_msal_app", return_value=mock_app):
status = get_auth_status()
assert status["logged_in"] is False
assert status["username"] == "user@example.com"
assert "auth login" in status["message"]


class TestGetToken:
def test_no_client_id_returns_none(self):
Expand Down
39 changes: 37 additions & 2 deletions tests/test_auth_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@

class TestSetup:
def test_setup_calls_login(self):
with patch("outpost.auth.login_interactive", return_value=True) as mock_login:
with patch("outpost.auth.login_browser", return_value=True) as mock_login:
result = runner.invoke(app, ["setup"], input="n\nUTC\n")
assert result.exit_code == 0
mock_login.assert_called_once()
Expand All @@ -26,7 +26,7 @@ def test_setup_no_client_id(self):
assert "not configured" in result.output.lower() or "Azure" in result.output

def test_setup_enable_teams(self):
with patch("outpost.auth.login_interactive", return_value=True) as mock_login, \
with patch("outpost.auth.login_browser", return_value=True) as mock_login, \
patch("outpost.config.save_config") as mock_save:
result = runner.invoke(app, ["setup"], input="y\nAmerica/New_York\n")
assert result.exit_code == 0
Expand Down Expand Up @@ -59,3 +59,38 @@ def test_status_json_output(self):
parsed = json.loads(result.output)
assert parsed["logged_in"] is True
assert parsed["username"] == "user@example.com"

def test_status_expired_shows_message(self):
status = {
"logged_in": False,
"username": "user@example.com",
"message": "Session expired for user@example.com — run 'outpost auth login' to re-authenticate.",
}
with patch("outpost.auth.get_auth_status", return_value=status):
result = runner.invoke(app, ["auth", "status"])
assert result.exit_code == 0
assert "expired" in result.output.lower()
assert "auth login" in result.output


class TestAuthLogin:
def test_login_browser_default(self):
with patch("outpost.auth.login_browser", return_value=True) as mock_browser, \
patch("outpost.auth.login_interactive") as mock_device:
result = runner.invoke(app, ["auth", "login"])
assert result.exit_code == 0
mock_browser.assert_called_once()
mock_device.assert_not_called()

def test_login_device_flag(self):
with patch("outpost.auth.login_interactive", return_value=True) as mock_device, \
patch("outpost.auth.login_browser") as mock_browser:
result = runner.invoke(app, ["auth", "login", "--device"])
assert result.exit_code == 0
mock_device.assert_called_once()
mock_browser.assert_not_called()

def test_login_failure_exits_nonzero(self):
with patch("outpost.auth.login_browser", return_value=False):
result = runner.invoke(app, ["auth", "login"])
assert result.exit_code == 1