#!/usr/bin/env python3
"""
Google OAuth2 Re-Authorization Script
Läuft auf Port 8080 und wartet auf den OAuth Callback.
Verwendung:
1. Auf VPS ausführen: python3 $HOME/aria/scripts/setup_google_auth.py
2. Den angezeigten Link im Browser öffnen (mit Google-Account einloggen)
3. Nach Bestätigung werden die Tokens automatisch gespeichert
"""
import json
import urllib.parse
import urllib.request
import sys
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
CREDS_FILE = '/root/.google_credentials.json'
REDIRECT_URI = 'http://localhost:8080/callback'
PORT = 8080
with open(CREDS_FILE) as f:
CREDS = json.load(f)
CLIENT_ID = CREDS['client_id']
CLIENT_SECRET = CREDS['client_secret']
SCOPES = ' '.join(CREDS['scopes'])
auth_code = None
server_done = threading.Event()
def build_auth_url():
params = {
'client_id': CLIENT_ID,
'redirect_uri': REDIRECT_URI,
'response_type': 'code',
'scope': SCOPES,
'access_type': 'offline',
'prompt': 'consent'
}
return 'https://accounts.google.com/o/oauth2/v2/auth?' + urllib.parse.urlencode(params)
def exchange_code(code):
data = urllib.parse.urlencode({
'code': code,
'client_id': CLIENT_ID,
'client_secret': CLIENT_SECRET,
'redirect_uri': REDIRECT_URI,
'grant_type': 'authorization_code'
}).encode()
req = urllib.request.Request(
'https://oauth2.googleapis.com/token',
data=data,
headers={'Content-Type': 'application/x-www-form-urlencoded'}
)
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
class CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
global auth_code
parsed = urllib.parse.urlparse(self.path)
if parsed.path == '/callback':
params = urllib.parse.parse_qs(parsed.query)
if 'code' in params:
auth_code = params['code'][0]
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(b'Aria: Google Auth erfolgreich!
Du kannst dieses Fenster schliessen.
')
server_done.set()
elif 'error' in params:
self.send_response(400)
self.end_headers()
self.wfile.write(f'Error: {params["error"]}'.encode())
server_done.set()
else:
self.send_response(404)
self.end_headers()
def log_message(self, format, *args):
pass # Silent
def main():
url = build_auth_url()
print('\n' + '='*60)
print('GOOGLE AUTH — SCHRITT 1:')
print('='*60)
print('\nOeffne diesen Link in deinem Browser:\n')
print(url)
print('\n' + '='*60)
print('Warte auf Callback... (Ctrl+C zum Abbrechen)')
print('='*60 + '\n')
server = HTTPServer(('0.0.0.0', PORT), CallbackHandler)
t = threading.Thread(target=server.serve_forever)
t.daemon = True
t.start()
server_done.wait(timeout=300) # 5 min timeout
server.shutdown()
if not auth_code:
print('ERROR: Kein Code erhalten (Timeout oder Fehler)')
sys.exit(1)
print('Code erhalten, tausche gegen Tokens...')
try:
result = exchange_code(auth_code)
except Exception as e:
print(f'ERROR beim Token-Austausch: {e}')
sys.exit(1)
with open(CREDS_FILE) as f:
creds = json.load(f)
creds['access_token'] = result['access_token']
if 'refresh_token' in result:
creds['refresh_token'] = result['refresh_token']
print('Neuer Refresh Token gespeichert.')
else:
print('Kein neuer Refresh Token (bestehender bleibt).')
with open(CREDS_FILE, 'w') as f:
json.dump(creds, f, indent=2)
print(f'\nERFOLG: Tokens gespeichert.')
print(f'Access Token gueltig fuer: ~{result.get("expires_in", 3600)//60} Min')
print('Der Cron-Job erneuert den Token ab jetzt automatisch alle 30 Min.')
if __name__ == '__main__':
main()