Store access tokens for multiple instances
This makes it so an app is created only once for each instance, instead of being re-created on each login. Prevents accumulations of authroized apps in https://mastodon.social/oauth/authorized_applications
This commit is contained in:
Ivan Habunek
parent
ed20c7fded
commit
3f44d560c8
|
|
@ -1,22 +1,18 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from toot import App, User, CLIENT_NAME, CLIENT_WEBSITE
|
||||
from toot.api import create_app, login, SCOPES
|
||||
|
||||
|
||||
class MockResponse:
|
||||
def __init__(self, response_data={}):
|
||||
self.response_data = response_data
|
||||
|
||||
def raise_for_status(self):
|
||||
pass
|
||||
|
||||
def json(self):
|
||||
return self.response_data
|
||||
from toot import App, CLIENT_NAME, CLIENT_WEBSITE
|
||||
from toot.api import create_app, login, SCOPES, AuthenticationError
|
||||
from tests.utils import MockResponse
|
||||
|
||||
|
||||
def test_create_app(monkeypatch):
|
||||
response = {
|
||||
'client_id': 'foo',
|
||||
'client_secret': 'bar',
|
||||
}
|
||||
|
||||
def mock_post(url, data):
|
||||
assert url == 'https://bigfish.software/api/v1/apps'
|
||||
assert data == {
|
||||
|
|
@ -25,24 +21,25 @@ def test_create_app(monkeypatch):
|
|||
'scopes': SCOPES,
|
||||
'redirect_uris': 'urn:ietf:wg:oauth:2.0:oob'
|
||||
}
|
||||
return MockResponse({
|
||||
'client_id': 'foo',
|
||||
'client_secret': 'bar',
|
||||
})
|
||||
return MockResponse(response)
|
||||
|
||||
monkeypatch.setattr(requests, 'post', mock_post)
|
||||
|
||||
app = create_app('https://bigfish.software')
|
||||
|
||||
assert isinstance(app, App)
|
||||
assert app.client_id == 'foo'
|
||||
assert app.client_secret == 'bar'
|
||||
assert create_app('bigfish.software') == response
|
||||
|
||||
|
||||
def test_login(monkeypatch):
|
||||
app = App('https://bigfish.software', 'foo', 'bar')
|
||||
app = App('bigfish.software', 'https://bigfish.software', 'foo', 'bar')
|
||||
|
||||
def mock_post(url, data):
|
||||
response = {
|
||||
'token_type': 'bearer',
|
||||
'scope': 'read write follow',
|
||||
'access_token': 'xxx',
|
||||
'created_at': 1492523699
|
||||
}
|
||||
|
||||
def mock_post(url, data, allow_redirects):
|
||||
assert not allow_redirects
|
||||
assert url == 'https://bigfish.software/oauth/token'
|
||||
assert data == {
|
||||
'grant_type': 'password',
|
||||
|
|
@ -52,14 +49,32 @@ def test_login(monkeypatch):
|
|||
'password': 'pass',
|
||||
'scope': SCOPES,
|
||||
}
|
||||
return MockResponse({
|
||||
'access_token': 'xxx',
|
||||
})
|
||||
|
||||
return MockResponse(response)
|
||||
|
||||
monkeypatch.setattr(requests, 'post', mock_post)
|
||||
|
||||
user = login(app, 'user', 'pass')
|
||||
assert login(app, 'user', 'pass') == response
|
||||
|
||||
assert isinstance(user, User)
|
||||
assert user.username == 'user'
|
||||
assert user.access_token == 'xxx'
|
||||
|
||||
def test_login_failed(monkeypatch):
|
||||
app = App('bigfish.software', 'https://bigfish.software', 'foo', 'bar')
|
||||
|
||||
def mock_post(url, data, allow_redirects):
|
||||
assert not allow_redirects
|
||||
assert url == 'https://bigfish.software/oauth/token'
|
||||
assert data == {
|
||||
'grant_type': 'password',
|
||||
'client_id': app.client_id,
|
||||
'client_secret': app.client_secret,
|
||||
'username': 'user',
|
||||
'password': 'pass',
|
||||
'scope': SCOPES,
|
||||
}
|
||||
|
||||
return MockResponse(is_redirect=True)
|
||||
|
||||
monkeypatch.setattr(requests, 'post', mock_post)
|
||||
|
||||
with pytest.raises(AuthenticationError):
|
||||
login(app, 'user', 'pass')
|
||||
|
|
|
|||
|
|
@ -7,8 +7,8 @@ from toot import console, User, App
|
|||
|
||||
from tests.utils import MockResponse
|
||||
|
||||
app = App('https://habunek.com', 'foo', 'bar')
|
||||
user = User('ivan@habunek.com', 'xxx')
|
||||
app = App('habunek.com', 'https://habunek.com', 'foo', 'bar')
|
||||
user = User('habunek.com', 'ivan@habunek.com', 'xxx')
|
||||
|
||||
|
||||
def uncolorize(text):
|
||||
|
|
@ -16,7 +16,7 @@ def uncolorize(text):
|
|||
return re.sub(r'\x1b[^m]*m', '', text)
|
||||
|
||||
|
||||
def test_print_usagecap(capsys):
|
||||
def test_print_usage(capsys):
|
||||
console.print_usage()
|
||||
out, err = capsys.readouterr()
|
||||
assert "toot - interact with Mastodon from the command line" in out
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
|
||||
class MockResponse:
|
||||
def __init__(self, response_data={}, ok=True):
|
||||
self.ok = ok
|
||||
def __init__(self, response_data={}, ok=True, is_redirect=False):
|
||||
self.response_data = response_data
|
||||
self.ok = ok
|
||||
self.is_redirect = is_redirect
|
||||
|
||||
def raise_for_status(self):
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -2,8 +2,8 @@
|
|||
|
||||
from collections import namedtuple
|
||||
|
||||
App = namedtuple('App', ['base_url', 'client_id', 'client_secret'])
|
||||
User = namedtuple('User', ['username', 'access_token'])
|
||||
App = namedtuple('App', ['instance', 'base_url', 'client_id', 'client_secret'])
|
||||
User = namedtuple('User', ['instance', 'username', 'access_token'])
|
||||
|
||||
DEFAULT_INSTANCE = 'mastodon.social'
|
||||
|
||||
|
|
|
|||
28
toot/api.py
28
toot/api.py
|
|
@ -20,6 +20,10 @@ class NotFoundError(ApiError):
|
|||
pass
|
||||
|
||||
|
||||
class AuthenticationError(ApiError):
|
||||
pass
|
||||
|
||||
|
||||
def _log_request(request):
|
||||
logger.debug(">>> \033[32m{} {}\033[0m".format(request.method, request.url))
|
||||
logger.debug(">>> HEADERS: \033[33m{}\033[0m".format(request.headers))
|
||||
|
|
@ -57,8 +61,6 @@ def _process_response(response):
|
|||
|
||||
raise ApiError(error)
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
return response.json()
|
||||
|
||||
|
||||
|
|
@ -88,7 +90,8 @@ def _post(app, user, url, data=None, files=None):
|
|||
return _process_response(response)
|
||||
|
||||
|
||||
def create_app(base_url):
|
||||
def create_app(instance):
|
||||
base_url = 'https://' + instance
|
||||
url = base_url + '/api/v1/apps'
|
||||
|
||||
response = requests.post(url, {
|
||||
|
|
@ -98,13 +101,7 @@ def create_app(base_url):
|
|||
'website': CLIENT_WEBSITE,
|
||||
})
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
client_id = data.get('client_id')
|
||||
client_secret = data.get('client_secret')
|
||||
|
||||
return App(base_url, client_id, client_secret)
|
||||
return _process_response(response)
|
||||
|
||||
|
||||
def login(app, username, password):
|
||||
|
|
@ -117,14 +114,13 @@ def login(app, username, password):
|
|||
'username': username,
|
||||
'password': password,
|
||||
'scope': SCOPES,
|
||||
})
|
||||
}, allow_redirects=False)
|
||||
|
||||
response.raise_for_status()
|
||||
# If auth fails, it redirects to the login page
|
||||
if response.is_redirect:
|
||||
raise AuthenticationError("Login failed")
|
||||
|
||||
data = response.json()
|
||||
access_token = data.get('access_token')
|
||||
|
||||
return User(username, access_token)
|
||||
return _process_response(response)
|
||||
|
||||
|
||||
def post_status(app, user, status, visibility='public', media_ids=None):
|
||||
|
|
|
|||
|
|
@ -4,11 +4,24 @@ import os
|
|||
|
||||
from . import User, App
|
||||
|
||||
# The dir where all toot configuration is stored
|
||||
CONFIG_DIR = os.environ['HOME'] + '/.config/toot/'
|
||||
CONFIG_APP_FILE = CONFIG_DIR + 'app.cfg'
|
||||
|
||||
# Subfolder where application access keys for various instances are stored
|
||||
INSTANCES_DIR = CONFIG_DIR + 'instances/'
|
||||
|
||||
# File in which user access token is stored
|
||||
CONFIG_USER_FILE = CONFIG_DIR + 'user.cfg'
|
||||
|
||||
|
||||
def get_instance_config_path(instance):
|
||||
return INSTANCES_DIR + instance
|
||||
|
||||
|
||||
def get_user_config_path():
|
||||
return CONFIG_USER_FILE
|
||||
|
||||
|
||||
def _load(file, tuple_class):
|
||||
if not os.path.exists(file):
|
||||
return None
|
||||
|
|
@ -28,28 +41,38 @@ def _save(file, named_tuple):
|
|||
|
||||
with open(file, 'w') as f:
|
||||
values = [v for v in named_tuple]
|
||||
return f.write("\n".join(values))
|
||||
f.write("\n".join(values))
|
||||
|
||||
|
||||
def load_app():
|
||||
return _load(CONFIG_APP_FILE, App)
|
||||
def load_app(instance):
|
||||
path = get_instance_config_path(instance)
|
||||
return _load(path, App)
|
||||
|
||||
|
||||
def load_user():
|
||||
return _load(CONFIG_USER_FILE, User)
|
||||
path = get_user_config_path()
|
||||
return _load(path, User)
|
||||
|
||||
|
||||
def save_app(app):
|
||||
return _save(CONFIG_APP_FILE, app)
|
||||
path = get_instance_config_path(app.instance)
|
||||
_save(path, app)
|
||||
return path
|
||||
|
||||
|
||||
def save_user(user):
|
||||
return _save(CONFIG_USER_FILE, user)
|
||||
path = get_user_config_path()
|
||||
_save(path, user)
|
||||
return path
|
||||
|
||||
|
||||
def delete_app(app):
|
||||
return os.unlink(CONFIG_APP_FILE)
|
||||
def delete_app(instance):
|
||||
path = get_instance_config_path(instance)
|
||||
os.unlink(path)
|
||||
return path
|
||||
|
||||
|
||||
def delete_user(user):
|
||||
return os.unlink(CONFIG_USER_FILE)
|
||||
def delete_user():
|
||||
path = get_user_config_path()
|
||||
os.unlink(path)
|
||||
return path
|
||||
|
|
|
|||
|
|
@ -15,9 +15,8 @@ from itertools import chain
|
|||
from argparse import ArgumentParser, FileType
|
||||
from textwrap import TextWrapper
|
||||
|
||||
from toot import api, DEFAULT_INSTANCE
|
||||
from toot import api, config, DEFAULT_INSTANCE, User, App
|
||||
from toot.api import ApiError
|
||||
from toot.config import save_user, load_user, load_app, save_app, CONFIG_APP_FILE, CONFIG_USER_FILE
|
||||
|
||||
|
||||
class ConsoleError(Exception):
|
||||
|
|
@ -44,38 +43,48 @@ def print_error(text):
|
|||
print(red(text), file=sys.stderr)
|
||||
|
||||
|
||||
def register_app(instance):
|
||||
print("Registering application with %s" % green(instance))
|
||||
|
||||
try:
|
||||
response = api.create_app(instance)
|
||||
except:
|
||||
raise ConsoleError("Registration failed. Did you enter a valid instance?")
|
||||
|
||||
base_url = 'https://' + instance
|
||||
|
||||
app = App(instance, base_url, response['client_id'], response['client_secret'])
|
||||
path = config.save_app(app)
|
||||
print("Application tokens saved to: {}".format(green(path)))
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def create_app_interactive():
|
||||
instance = input("Choose an instance [%s]: " % green(DEFAULT_INSTANCE))
|
||||
if not instance:
|
||||
instance = DEFAULT_INSTANCE
|
||||
|
||||
base_url = 'https://{}'.format(instance)
|
||||
|
||||
print("Registering application with %s" % green(base_url))
|
||||
try:
|
||||
app = api.create_app(base_url)
|
||||
except:
|
||||
raise ConsoleError("Failed authenticating application. Did you enter a valid instance?")
|
||||
|
||||
save_app(app)
|
||||
print("Application tokens saved to: {}".format(green(CONFIG_APP_FILE)))
|
||||
|
||||
return app
|
||||
return config.load_app(instance) or register_app(instance)
|
||||
|
||||
|
||||
def login_interactive(app):
|
||||
print("\nLog in to " + green(app.base_url))
|
||||
print("\nLog in to " + green(app.instance))
|
||||
email = input('Email: ')
|
||||
password = getpass('Password: ')
|
||||
|
||||
print("Authenticating...")
|
||||
if not email or not password:
|
||||
raise ConsoleError("Email and password cannot be empty.")
|
||||
|
||||
try:
|
||||
user = api.login(app, email, password)
|
||||
except:
|
||||
print("Authenticating...")
|
||||
response = api.login(app, email, password)
|
||||
except ApiError:
|
||||
raise ConsoleError("Login failed")
|
||||
|
||||
save_user(user)
|
||||
print("User token saved to " + green(CONFIG_USER_FILE))
|
||||
user = User(app.instance, email, response['access_token'])
|
||||
path = config.save_user(user)
|
||||
print("Access token saved to: " + green(path))
|
||||
|
||||
return user
|
||||
|
||||
|
|
@ -193,10 +202,9 @@ def cmd_auth(app, user, args):
|
|||
parser.parse_args(args)
|
||||
|
||||
if app and user:
|
||||
print("You are logged in to " + green(app.base_url))
|
||||
print("Username: " + green(user.username))
|
||||
print("App data: " + green(CONFIG_APP_FILE))
|
||||
print("User data: " + green(CONFIG_USER_FILE))
|
||||
print("You are logged in to {} as {}".format(green(app.instance), green(user.username)))
|
||||
print("User data: " + green(config.get_user_config_path()))
|
||||
print("App data: " + green(config.get_instance_config_path(app.instance)))
|
||||
else:
|
||||
print("You are not logged in")
|
||||
|
||||
|
|
@ -219,9 +227,9 @@ def cmd_logout(app, user, args):
|
|||
epilog="https://github.com/ihabunek/toot")
|
||||
parser.parse_args(args)
|
||||
|
||||
os.unlink(CONFIG_APP_FILE)
|
||||
os.unlink(CONFIG_USER_FILE)
|
||||
print("You are now logged out")
|
||||
config.delete_user()
|
||||
|
||||
print(green("✓ You are now logged out"))
|
||||
|
||||
|
||||
def cmd_upload(app, user, args):
|
||||
|
|
@ -348,8 +356,8 @@ def cmd_whoami(app, user, args):
|
|||
|
||||
|
||||
def run_command(command, args):
|
||||
app = load_app()
|
||||
user = load_user()
|
||||
user = config.load_user()
|
||||
app = config.load_app(user.instance) if user else None
|
||||
|
||||
# Commands which can run when not logged in
|
||||
if command == 'login':
|
||||
|
|
|
|||
Loading…
Reference in New Issue