365 lines
12 KiB
Python
Executable file
365 lines
12 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
Microsoft Graph API Client for Mikkel (mikpar@outlook.dk)
|
|
|
|
Usage:
|
|
python3 graph_client.py auth # Authenticate
|
|
python3 graph_client.py mail list --limit 10 # List emails
|
|
python3 graph_client.py mail read --id <id> # Read email
|
|
python3 graph_client.py mail send --to x@y.com --subject "Hi" --body "Hello"
|
|
python3 graph_client.py drive list # List OneDrive root
|
|
python3 graph_client.py drive list --path "Documents" # List folder
|
|
python3 graph_client.py drive download --path "file.pdf" --output ./file.pdf
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
|
|
# Configuration
|
|
CLIENT_ID = "96861287-203e-4d84-ad3b-379d63ba5eb2"
|
|
TENANT_ID = "3a22994f-512f-447e-b8fa-19bbc849bd0c"
|
|
CLIENT_SECRET = "kiB8Q~TmhWDKHRdLqBi_RqRIIAjTkljyL6xE0aHU"
|
|
SCOPES = "Mail.Read Mail.Send Files.ReadWrite User.Read offline_access"
|
|
|
|
# Use 'consumers' endpoint for personal Microsoft accounts
|
|
AUTH_URL = "https://login.microsoftonline.com/consumers/oauth2/v2.0"
|
|
GRAPH_URL = "https://graph.microsoft.com/v1.0"
|
|
|
|
TOKEN_FILE = os.path.expanduser("~/.config/microsoft-graph/token.json")
|
|
|
|
|
|
def save_token(token_data):
|
|
"""Save token to file."""
|
|
os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True)
|
|
token_data["saved_at"] = int(time.time())
|
|
with open(TOKEN_FILE, "w") as f:
|
|
json.dump(token_data, f, indent=2)
|
|
|
|
|
|
def load_token():
|
|
"""Load token from file."""
|
|
if not os.path.exists(TOKEN_FILE):
|
|
return None
|
|
with open(TOKEN_FILE, "r") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def refresh_token(refresh_token_str):
|
|
"""Refresh the access token."""
|
|
data = urlencode({
|
|
"client_id": CLIENT_ID,
|
|
"client_secret": CLIENT_SECRET,
|
|
"refresh_token": refresh_token_str,
|
|
"grant_type": "refresh_token",
|
|
"scope": SCOPES,
|
|
}).encode()
|
|
|
|
req = Request(f"{AUTH_URL}/token", data=data, method="POST")
|
|
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
|
|
try:
|
|
with urlopen(req, timeout=30) as resp:
|
|
token_data = json.loads(resp.read().decode())
|
|
save_token(token_data)
|
|
return token_data
|
|
except HTTPError as e:
|
|
print(f"Token refresh failed: {e.code}")
|
|
return None
|
|
|
|
|
|
def get_access_token():
|
|
"""Get valid access token, refreshing if needed."""
|
|
token_data = load_token()
|
|
if not token_data:
|
|
print("No token found. Run 'auth' command first.")
|
|
sys.exit(1)
|
|
|
|
# Check if token is expired (with 5 min buffer)
|
|
saved_at = token_data.get("saved_at", 0)
|
|
expires_in = token_data.get("expires_in", 3600)
|
|
if time.time() > saved_at + expires_in - 300:
|
|
print("Token expired, refreshing...")
|
|
token_data = refresh_token(token_data.get("refresh_token"))
|
|
if not token_data:
|
|
print("Token refresh failed. Run 'auth' command.")
|
|
sys.exit(1)
|
|
|
|
return token_data["access_token"]
|
|
|
|
|
|
def device_code_auth():
|
|
"""Authenticate using device code flow."""
|
|
# Request device code
|
|
data = urlencode({
|
|
"client_id": CLIENT_ID,
|
|
"scope": SCOPES,
|
|
}).encode()
|
|
|
|
req = Request(f"{AUTH_URL}/devicecode", data=data, method="POST")
|
|
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
|
|
try:
|
|
with urlopen(req, timeout=30) as resp:
|
|
device_data = json.loads(resp.read().decode())
|
|
except HTTPError as e:
|
|
return {"error": f"Device code request failed: {e.code}"}
|
|
|
|
print("\n" + "=" * 50)
|
|
print("📱 AUTHENTICATION REQUIRED")
|
|
print("=" * 50)
|
|
print(f"\n1. Open: {device_data['verification_uri']}")
|
|
print(f"2. Enter code: {device_data['user_code']}")
|
|
print(f"3. Sign in with mikpar@outlook.dk")
|
|
print("\nWaiting for authorization...")
|
|
|
|
# Poll for token
|
|
interval = device_data.get("interval", 5)
|
|
expires_in = device_data.get("expires_in", 900)
|
|
start_time = time.time()
|
|
|
|
while time.time() - start_time < expires_in:
|
|
time.sleep(interval)
|
|
|
|
poll_data = urlencode({
|
|
"client_id": CLIENT_ID,
|
|
"client_secret": CLIENT_SECRET,
|
|
"device_code": device_data["device_code"],
|
|
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
|
|
}).encode()
|
|
|
|
poll_req = Request(f"{AUTH_URL}/token", data=poll_data, method="POST")
|
|
poll_req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
|
|
try:
|
|
with urlopen(poll_req, timeout=30) as resp:
|
|
token_data = json.loads(resp.read().decode())
|
|
save_token(token_data)
|
|
print("\n✅ Authentication successful! Token saved.")
|
|
return {"status": "authenticated"}
|
|
except HTTPError as e:
|
|
error_body = json.loads(e.read().decode())
|
|
error_code = error_body.get("error")
|
|
if error_code == "authorization_pending":
|
|
continue
|
|
elif error_code == "slow_down":
|
|
interval += 5
|
|
continue
|
|
else:
|
|
return {"error": f"Auth failed: {error_body.get('error_description')}"}
|
|
|
|
return {"error": "Authentication timed out"}
|
|
|
|
|
|
def graph_request(endpoint, method="GET", data=None, headers=None):
|
|
"""Make a Graph API request."""
|
|
token = get_access_token()
|
|
url = f"{GRAPH_URL}{endpoint}"
|
|
|
|
req_headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
if headers:
|
|
req_headers.update(headers)
|
|
|
|
req = Request(url, method=method, headers=req_headers)
|
|
if data:
|
|
req.data = json.dumps(data).encode()
|
|
|
|
try:
|
|
with urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except HTTPError as e:
|
|
error_body = e.read().decode()
|
|
try:
|
|
return {"error": json.loads(error_body)}
|
|
except:
|
|
return {"error": f"HTTP {e.code}: {error_body[:200]}"}
|
|
|
|
|
|
def list_mail(limit=10, folder="inbox"):
|
|
"""List emails."""
|
|
endpoint = f"/me/mailFolders/{folder}/messages?$top={limit}&$orderby=receivedDateTime desc"
|
|
endpoint += "&$select=id,subject,from,receivedDateTime,isRead,bodyPreview"
|
|
result = graph_request(endpoint)
|
|
|
|
if "error" in result:
|
|
return result
|
|
|
|
messages = []
|
|
for msg in result.get("value", []):
|
|
messages.append({
|
|
"id": msg["id"],
|
|
"subject": msg.get("subject", "(no subject)"),
|
|
"from": msg.get("from", {}).get("emailAddress", {}).get("address", "unknown"),
|
|
"date": msg.get("receivedDateTime"),
|
|
"read": msg.get("isRead", False),
|
|
"preview": msg.get("bodyPreview", "")[:100]
|
|
})
|
|
|
|
return {"count": len(messages), "messages": messages}
|
|
|
|
|
|
def read_mail(msg_id):
|
|
"""Read a specific email."""
|
|
endpoint = f"/me/messages/{msg_id}"
|
|
result = graph_request(endpoint)
|
|
|
|
if "error" in result:
|
|
return result
|
|
|
|
return {
|
|
"id": result["id"],
|
|
"subject": result.get("subject"),
|
|
"from": result.get("from", {}).get("emailAddress", {}),
|
|
"to": [r.get("emailAddress", {}) for r in result.get("toRecipients", [])],
|
|
"date": result.get("receivedDateTime"),
|
|
"body": result.get("body", {}).get("content", "")[:5000]
|
|
}
|
|
|
|
|
|
def send_mail(to, subject, body, cc=None):
|
|
"""Send an email."""
|
|
message = {
|
|
"message": {
|
|
"subject": subject,
|
|
"body": {
|
|
"contentType": "Text",
|
|
"content": body
|
|
},
|
|
"toRecipients": [{"emailAddress": {"address": to}}]
|
|
}
|
|
}
|
|
|
|
if cc:
|
|
message["message"]["ccRecipients"] = [{"emailAddress": {"address": cc}}]
|
|
|
|
result = graph_request("/me/sendMail", method="POST", data=message)
|
|
|
|
if result is None:
|
|
return {"status": "sent", "to": to, "subject": subject}
|
|
return result
|
|
|
|
|
|
def list_drive(path=None):
|
|
"""List OneDrive contents."""
|
|
if path:
|
|
endpoint = f"/me/drive/root:/{path}:/children"
|
|
else:
|
|
endpoint = "/me/drive/root/children"
|
|
|
|
endpoint += "?$select=id,name,size,folder,file,lastModifiedDateTime"
|
|
result = graph_request(endpoint)
|
|
|
|
if "error" in result:
|
|
return result
|
|
|
|
items = []
|
|
for item in result.get("value", []):
|
|
items.append({
|
|
"id": item["id"],
|
|
"name": item["name"],
|
|
"type": "folder" if "folder" in item else "file",
|
|
"size": item.get("size"),
|
|
"modified": item.get("lastModifiedDateTime")
|
|
})
|
|
|
|
return {"count": len(items), "items": items}
|
|
|
|
|
|
def download_file(path, output):
|
|
"""Download a file from OneDrive."""
|
|
# Get download URL
|
|
endpoint = f"/me/drive/root:/{path}"
|
|
result = graph_request(endpoint)
|
|
|
|
if "error" in result:
|
|
return result
|
|
|
|
download_url = result.get("@microsoft.graph.downloadUrl")
|
|
if not download_url:
|
|
return {"error": "No download URL available"}
|
|
|
|
# Download file
|
|
try:
|
|
req = Request(download_url)
|
|
with urlopen(req, timeout=60) as resp:
|
|
with open(output, "wb") as f:
|
|
f.write(resp.read())
|
|
return {"status": "downloaded", "path": path, "output": output, "size": os.path.getsize(output)}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Microsoft Graph API Client")
|
|
subparsers = parser.add_subparsers(dest="command", help="Commands")
|
|
|
|
# Auth
|
|
subparsers.add_parser("auth", help="Authenticate with device code flow")
|
|
|
|
# Mail commands
|
|
mail_parser = subparsers.add_parser("mail", help="Email operations")
|
|
mail_sub = mail_parser.add_subparsers(dest="mail_cmd")
|
|
|
|
mail_list = mail_sub.add_parser("list", help="List emails")
|
|
mail_list.add_argument("--limit", type=int, default=10)
|
|
mail_list.add_argument("--folder", default="inbox")
|
|
|
|
mail_read = mail_sub.add_parser("read", help="Read email")
|
|
mail_read.add_argument("--id", required=True)
|
|
|
|
mail_send = mail_sub.add_parser("send", help="Send email")
|
|
mail_send.add_argument("--to", required=True)
|
|
mail_send.add_argument("--subject", required=True)
|
|
mail_send.add_argument("--body", required=True)
|
|
mail_send.add_argument("--cc")
|
|
|
|
# Drive commands
|
|
drive_parser = subparsers.add_parser("drive", help="OneDrive operations")
|
|
drive_sub = drive_parser.add_subparsers(dest="drive_cmd")
|
|
|
|
drive_list = drive_sub.add_parser("list", help="List files/folders")
|
|
drive_list.add_argument("--path")
|
|
|
|
drive_dl = drive_sub.add_parser("download", help="Download file")
|
|
drive_dl.add_argument("--path", required=True)
|
|
drive_dl.add_argument("--output", required=True)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "auth":
|
|
result = device_code_auth()
|
|
elif args.command == "mail":
|
|
if args.mail_cmd == "list":
|
|
result = list_mail(args.limit, args.folder)
|
|
elif args.mail_cmd == "read":
|
|
result = read_mail(args.id)
|
|
elif args.mail_cmd == "send":
|
|
result = send_mail(args.to, args.subject, args.body, args.cc)
|
|
else:
|
|
mail_parser.print_help()
|
|
sys.exit(1)
|
|
elif args.command == "drive":
|
|
if args.drive_cmd == "list":
|
|
result = list_drive(args.path)
|
|
elif args.drive_cmd == "download":
|
|
result = download_file(args.path, args.output)
|
|
else:
|
|
drive_parser.print_help()
|
|
sys.exit(1)
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|