Move some more methods, fix crypto heckup
This commit is contained in:
halcy
parent
48f1d31c72
commit
d9cd7547fd
@ -24,6 +24,7 @@ from .compat import urlparse
|
|||||||
|
|
||||||
from .utility import parse_version_string, max_version, api_version
|
from .utility import parse_version_string, max_version, api_version
|
||||||
from .utility import AttribAccessDict, AttribAccessDict
|
from .utility import AttribAccessDict, AttribAccessDict
|
||||||
|
from .utility import Mastodon as Utility
|
||||||
|
|
||||||
from .error import *
|
from .error import *
|
||||||
from .versions import _DICT_VERSION_APPLICATION, _DICT_VERSION_MENTION, _DICT_VERSION_MEDIA, _DICT_VERSION_ACCOUNT, _DICT_VERSION_POLL, \
|
from .versions import _DICT_VERSION_APPLICATION, _DICT_VERSION_MENTION, _DICT_VERSION_MEDIA, _DICT_VERSION_ACCOUNT, _DICT_VERSION_POLL, \
|
||||||
@ -39,12 +40,14 @@ from .defaults import _DEFAULT_TIMEOUT, _DEFAULT_SCOPES, _DEFAULT_STREAM_TIMEOUT
|
|||||||
from .defaults import _SCOPE_SETS
|
from .defaults import _SCOPE_SETS
|
||||||
|
|
||||||
from .internals import Mastodon as Internals
|
from .internals import Mastodon as Internals
|
||||||
|
from .authentication import Mastodon as Authentication
|
||||||
from .accounts import Mastodon as Accounts
|
from .accounts import Mastodon as Accounts
|
||||||
|
from .instance import Mastodon as Instance
|
||||||
|
|
||||||
##
|
##
|
||||||
# The actual Mastodon class
|
# The actual Mastodon class
|
||||||
###
|
###
|
||||||
class Mastodon(Internals, Accounts):
|
class Mastodon(Utility, Authentication, Accounts, Instance):
|
||||||
"""
|
"""
|
||||||
Thorough and easy to use Mastodon
|
Thorough and easy to use Mastodon
|
||||||
API wrapper in Python.
|
API wrapper in Python.
|
||||||
@ -54,424 +57,6 @@ class Mastodon(Internals, Accounts):
|
|||||||
# Support level
|
# Support level
|
||||||
__SUPPORTED_MASTODON_VERSION = "3.5.5"
|
__SUPPORTED_MASTODON_VERSION = "3.5.5"
|
||||||
|
|
||||||
###
|
|
||||||
# Registering apps
|
|
||||||
###
|
|
||||||
@staticmethod
|
|
||||||
def create_app(client_name, scopes=_DEFAULT_SCOPES, redirect_uris=None, website=None, to_file=None,
|
|
||||||
api_base_url=None, request_timeout=_DEFAULT_TIMEOUT, session=None):
|
|
||||||
"""
|
|
||||||
Create a new app with given `client_name` and `scopes` (The basic scopes are "read", "write", "follow" and "push"
|
|
||||||
- more granular scopes are available, please refer to Mastodon documentation for which) on the instance given
|
|
||||||
by `api_base_url`.
|
|
||||||
|
|
||||||
Specify `redirect_uris` if you want users to be redirected to a certain page after authenticating in an OAuth flow.
|
|
||||||
You can specify multiple URLs by passing a list. Note that if you wish to use OAuth authentication with redirects,
|
|
||||||
the redirect URI must be one of the URLs specified here.
|
|
||||||
|
|
||||||
Specify `to_file` to persist your app's info to a file so you can use it in the constructor.
|
|
||||||
Specify `website` to give a website for your app.
|
|
||||||
|
|
||||||
Specify `session` with a requests.Session for it to be used instead of the default. This can be
|
|
||||||
used to, amongst other things, adjust proxy or SSL certificate settings.
|
|
||||||
|
|
||||||
Presently, app registration is open by default, but this is not guaranteed to be the case for all
|
|
||||||
Mastodon instances in the future.
|
|
||||||
|
|
||||||
|
|
||||||
Returns `client_id` and `client_secret`, both as strings.
|
|
||||||
"""
|
|
||||||
if api_base_url is None:
|
|
||||||
raise MastodonIllegalArgumentError("API base URL is required.")
|
|
||||||
api_base_url = Mastodon.__protocolize(api_base_url)
|
|
||||||
|
|
||||||
request_data = {
|
|
||||||
'client_name': client_name,
|
|
||||||
'scopes': " ".join(scopes)
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
if redirect_uris is not None:
|
|
||||||
if isinstance(redirect_uris, (list, tuple)):
|
|
||||||
redirect_uris = "\n".join(list(redirect_uris))
|
|
||||||
request_data['redirect_uris'] = redirect_uris
|
|
||||||
else:
|
|
||||||
request_data['redirect_uris'] = 'urn:ietf:wg:oauth:2.0:oob'
|
|
||||||
if website is not None:
|
|
||||||
request_data['website'] = website
|
|
||||||
if session:
|
|
||||||
ret = session.post(api_base_url + '/api/v1/apps', data=request_data, timeout=request_timeout)
|
|
||||||
response = ret.json()
|
|
||||||
else:
|
|
||||||
response = requests.post(api_base_url + '/api/v1/apps', data=request_data, timeout=request_timeout)
|
|
||||||
response = response.json()
|
|
||||||
except Exception as e:
|
|
||||||
raise MastodonNetworkError("Could not complete request: %s" % e)
|
|
||||||
|
|
||||||
if to_file is not None:
|
|
||||||
with open(to_file, 'w') as secret_file:
|
|
||||||
secret_file.write(response['client_id'] + "\n")
|
|
||||||
secret_file.write(response['client_secret'] + "\n")
|
|
||||||
secret_file.write(api_base_url + "\n")
|
|
||||||
secret_file.write(client_name + "\n")
|
|
||||||
|
|
||||||
return (response['client_id'], response['client_secret'])
|
|
||||||
|
|
||||||
###
|
|
||||||
# Authentication, including constructor
|
|
||||||
###
|
|
||||||
def __init__(self, client_id=None, client_secret=None, access_token=None, api_base_url=None, debug_requests=False,
|
|
||||||
ratelimit_method="wait", ratelimit_pacefactor=1.1, request_timeout=_DEFAULT_TIMEOUT, mastodon_version=None,
|
|
||||||
version_check_mode="created", session=None, feature_set="mainline", user_agent="mastodonpy", lang=None):
|
|
||||||
"""
|
|
||||||
Create a new API wrapper instance based on the given `client_secret` and `client_id` on the
|
|
||||||
instance given by `api_base_url`. If you give a `client_id` and it is not a file, you must
|
|
||||||
also give a secret. If you specify an `access_token` then you don't need to specify a `client_id`.
|
|
||||||
It is allowed to specify neither - in this case, you will be restricted to only using endpoints
|
|
||||||
that do not require authentication. If a file is given as `client_id`, client ID, secret and
|
|
||||||
base url are read from that file.
|
|
||||||
|
|
||||||
You can also specify an `access_token`, directly or as a file (as written by :ref:`log_in() <log_in()>`). If
|
|
||||||
a file is given, Mastodon.py also tries to load the base URL from this file, if present. A
|
|
||||||
client id and secret are not required in this case.
|
|
||||||
|
|
||||||
Mastodon.py can try to respect rate limits in several ways, controlled by `ratelimit_method`.
|
|
||||||
"throw" makes functions throw a `MastodonRatelimitError` when the rate
|
|
||||||
limit is hit. "wait" mode will, once the limit is hit, wait and retry the request as soon
|
|
||||||
as the rate limit resets, until it succeeds. "pace" works like throw, but tries to wait in
|
|
||||||
between calls so that the limit is generally not hit (how hard it tries to avoid hitting the rate
|
|
||||||
limit can be controlled by ratelimit_pacefactor). The default setting is "wait". Note that
|
|
||||||
even in "wait" and "pace" mode, requests can still fail due to network or other problems! Also
|
|
||||||
note that "pace" and "wait" are NOT thread safe.
|
|
||||||
|
|
||||||
By default, a timeout of 300 seconds is used for all requests. If you wish to change this,
|
|
||||||
pass the desired timeout (in seconds) as `request_timeout`.
|
|
||||||
|
|
||||||
For fine-tuned control over the requests object use `session` with a requests.Session.
|
|
||||||
|
|
||||||
The `mastodon_version` parameter can be used to specify the version of Mastodon that Mastodon.py will
|
|
||||||
expect to be installed on the server. The function will throw an error if an unparseable
|
|
||||||
Version is specified. If no version is specified, Mastodon.py will set `mastodon_version` to the
|
|
||||||
detected version.
|
|
||||||
|
|
||||||
The version check mode can be set to "created" (the default behaviour), "changed" or "none". If set to
|
|
||||||
"created", Mastodon.py will throw an error if the version of Mastodon it is connected to is too old
|
|
||||||
to have an endpoint. If it is set to "changed", it will throw an error if the endpoint's behaviour has
|
|
||||||
changed after the version of Mastodon that is connected has been released. If it is set to "none",
|
|
||||||
version checking is disabled.
|
|
||||||
|
|
||||||
`feature_set` can be used to enable behaviour specific to non-mainline Mastodon API implementations.
|
|
||||||
Details are documented in the functions that provide such functionality. Currently supported feature
|
|
||||||
sets are `mainline`, `fedibird` and `pleroma`.
|
|
||||||
|
|
||||||
For some Mastodon instances a `User-Agent` header is needed. This can be set by parameter `user_agent`. Starting from
|
|
||||||
Mastodon.py 1.5.2 `create_app()` stores the application name into the client secret file. If `client_id` points to this file,
|
|
||||||
the app name will be used as `User-Agent` header as default. It is possible to modify old secret files and append
|
|
||||||
a client app name to use it as a `User-Agent` name.
|
|
||||||
|
|
||||||
`lang` can be used to change the locale Mastodon will use to generate responses. Valid parameters are all ISO 639-1 (two letter)
|
|
||||||
or for a language that has none, 639-3 (three letter) language codes. This affects some error messages (those related to validation) and
|
|
||||||
trends. You can change the language using :ref:`set_language()`.
|
|
||||||
|
|
||||||
If no other `User-Agent` is specified, "mastodonpy" will be used.
|
|
||||||
"""
|
|
||||||
self.api_base_url = api_base_url
|
|
||||||
if self.api_base_url is not None:
|
|
||||||
self.api_base_url = self.__protocolize(self.api_base_url)
|
|
||||||
self.client_id = client_id
|
|
||||||
self.client_secret = client_secret
|
|
||||||
self.access_token = access_token
|
|
||||||
self.debug_requests = debug_requests
|
|
||||||
self.ratelimit_method = ratelimit_method
|
|
||||||
self._token_expired = datetime.datetime.now()
|
|
||||||
self._refresh_token = None
|
|
||||||
|
|
||||||
self.__logged_in_id = None
|
|
||||||
|
|
||||||
self.ratelimit_limit = 300
|
|
||||||
self.ratelimit_reset = time.time()
|
|
||||||
self.ratelimit_remaining = 300
|
|
||||||
self.ratelimit_lastcall = time.time()
|
|
||||||
self.ratelimit_pacefactor = ratelimit_pacefactor
|
|
||||||
|
|
||||||
self.request_timeout = request_timeout
|
|
||||||
|
|
||||||
if session:
|
|
||||||
self.session = session
|
|
||||||
else:
|
|
||||||
self.session = requests.Session()
|
|
||||||
|
|
||||||
self.feature_set = feature_set
|
|
||||||
if not self.feature_set in ["mainline", "fedibird", "pleroma"]:
|
|
||||||
raise MastodonIllegalArgumentError('Requested invalid feature set')
|
|
||||||
|
|
||||||
# General defined user-agent
|
|
||||||
self.user_agent = user_agent
|
|
||||||
|
|
||||||
# Save language
|
|
||||||
self.lang = lang
|
|
||||||
|
|
||||||
# Token loading
|
|
||||||
if self.client_id is not None:
|
|
||||||
if os.path.isfile(self.client_id):
|
|
||||||
with open(self.client_id, 'r') as secret_file:
|
|
||||||
self.client_id = secret_file.readline().rstrip()
|
|
||||||
self.client_secret = secret_file.readline().rstrip()
|
|
||||||
|
|
||||||
try_base_url = secret_file.readline().rstrip()
|
|
||||||
if try_base_url is not None and len(try_base_url) != 0:
|
|
||||||
try_base_url = Mastodon.__protocolize(try_base_url)
|
|
||||||
if not (self.api_base_url is None or try_base_url == self.api_base_url):
|
|
||||||
raise MastodonIllegalArgumentError('Mismatch in base URLs between files and/or specified')
|
|
||||||
self.api_base_url = try_base_url
|
|
||||||
|
|
||||||
# With new registrations we support the 4th line to store a client_name and use it as user-agent
|
|
||||||
client_name = secret_file.readline()
|
|
||||||
if client_name and self.user_agent is None:
|
|
||||||
self.user_agent = client_name.rstrip()
|
|
||||||
else:
|
|
||||||
if self.client_secret is None:
|
|
||||||
raise MastodonIllegalArgumentError('Specified client id directly, but did not supply secret')
|
|
||||||
|
|
||||||
if self.access_token is not None and os.path.isfile(self.access_token):
|
|
||||||
with open(self.access_token, 'r') as token_file:
|
|
||||||
self.access_token = token_file.readline().rstrip()
|
|
||||||
|
|
||||||
# For newer versions, we also store the URL
|
|
||||||
try_base_url = token_file.readline().rstrip()
|
|
||||||
if try_base_url is not None and len(try_base_url) != 0:
|
|
||||||
try_base_url = Mastodon.__protocolize(try_base_url)
|
|
||||||
if not (self.api_base_url is None or try_base_url == self.api_base_url):
|
|
||||||
raise MastodonIllegalArgumentError('Mismatch in base URLs between files and/or specified')
|
|
||||||
self.api_base_url = try_base_url
|
|
||||||
|
|
||||||
# For EVEN newer vesions, we ALSO ALSO store the client id and secret so that you don't need to reauth to revoke
|
|
||||||
if self.client_id is None:
|
|
||||||
try:
|
|
||||||
self.client_id = token_file.readline().rstrip()
|
|
||||||
self.client_secret = token_file.readline().rstrip()
|
|
||||||
except:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# Verify we have a base URL, protocolize
|
|
||||||
if self.api_base_url is None:
|
|
||||||
raise MastodonIllegalArgumentError("API base URL is required.")
|
|
||||||
self.api_base_url = Mastodon.__protocolize(self.api_base_url)
|
|
||||||
|
|
||||||
if not version_check_mode in ["created", "changed", "none"]:
|
|
||||||
raise MastodonIllegalArgumentError("Invalid version check method.")
|
|
||||||
self.version_check_mode = version_check_mode
|
|
||||||
|
|
||||||
self.mastodon_major = 1
|
|
||||||
self.mastodon_minor = 0
|
|
||||||
self.mastodon_patch = 0
|
|
||||||
self.version_check_worked = None
|
|
||||||
|
|
||||||
# Versioning
|
|
||||||
if mastodon_version is None and self.version_check_mode != 'none':
|
|
||||||
self.retrieve_mastodon_version()
|
|
||||||
elif self.version_check_mode != 'none':
|
|
||||||
try:
|
|
||||||
self.mastodon_major, self.mastodon_minor, self.mastodon_patch = parse_version_string(mastodon_version)
|
|
||||||
except:
|
|
||||||
raise MastodonVersionError("Bad version specified")
|
|
||||||
|
|
||||||
# Ratelimiting parameter check
|
|
||||||
if ratelimit_method not in ["throw", "wait", "pace"]:
|
|
||||||
raise MastodonIllegalArgumentError("Invalid ratelimit method.")
|
|
||||||
|
|
||||||
def auth_request_url(self, client_id=None, redirect_uris="urn:ietf:wg:oauth:2.0:oob", scopes=_DEFAULT_SCOPES, force_login=False, state=None, lang=None):
|
|
||||||
"""
|
|
||||||
Returns the URL that a client needs to request an OAuth grant from the server.
|
|
||||||
|
|
||||||
To log in with OAuth, send your user to this URL. The user will then log in and
|
|
||||||
get a code which you can pass to :ref:`log_in() <log_in()>`.
|
|
||||||
|
|
||||||
`scopes` are as in :ref:`log_in() <log_in()>`, redirect_uris is where the user should be redirected to
|
|
||||||
after authentication. Note that `redirect_uris` must be one of the URLs given during
|
|
||||||
app registration. When using urn:ietf:wg:oauth:2.0:oob, the code is simply displayed,
|
|
||||||
otherwise it is added to the given URL as the "code" request parameter.
|
|
||||||
|
|
||||||
Pass force_login if you want the user to always log in even when already logged
|
|
||||||
into web Mastodon (i.e. when registering multiple different accounts in an app).
|
|
||||||
|
|
||||||
`state` is the oauth `state` parameter to pass to the server. It is strongly suggested
|
|
||||||
to use a random, nonguessable value (i.e. nothing meaningful and no incrementing ID)
|
|
||||||
to preserve security guarantees. It can be left out for non-web login flows.
|
|
||||||
|
|
||||||
Pass an ISO 639-1 (two letter) or, for languages that do not have one, 639-3 (three letter)
|
|
||||||
language code as `lang` to control the display language for the oauth form.
|
|
||||||
"""
|
|
||||||
if client_id is None:
|
|
||||||
client_id = self.client_id
|
|
||||||
else:
|
|
||||||
if os.path.isfile(client_id):
|
|
||||||
with open(client_id, 'r') as secret_file:
|
|
||||||
client_id = secret_file.readline().rstrip()
|
|
||||||
|
|
||||||
params = dict()
|
|
||||||
params['client_id'] = client_id
|
|
||||||
params['response_type'] = "code"
|
|
||||||
params['redirect_uri'] = redirect_uris
|
|
||||||
params['scope'] = " ".join(scopes)
|
|
||||||
params['force_login'] = force_login
|
|
||||||
params['state'] = state
|
|
||||||
params['lang'] = lang
|
|
||||||
formatted_params = urlencode(params)
|
|
||||||
return "".join([self.api_base_url, "/oauth/authorize?", formatted_params])
|
|
||||||
|
|
||||||
def log_in(self, username=None, password=None, code=None, redirect_uri="urn:ietf:wg:oauth:2.0:oob", refresh_token=None, scopes=_DEFAULT_SCOPES, to_file=None):
|
|
||||||
"""
|
|
||||||
Get the access token for a user.
|
|
||||||
|
|
||||||
The username is the email address used to log in into Mastodon.
|
|
||||||
|
|
||||||
Can persist access token to file `to_file`, to be used in the constructor.
|
|
||||||
|
|
||||||
Handles password and OAuth-based authorization.
|
|
||||||
|
|
||||||
Will throw a `MastodonIllegalArgumentError` if the OAuth flow data or the
|
|
||||||
username / password credentials given are incorrect, and
|
|
||||||
`MastodonAPIError` if all of the requested scopes were not granted.
|
|
||||||
|
|
||||||
For OAuth 2, obtain a code via having your user go to the URL returned by
|
|
||||||
:ref:`auth_request_url() <auth_request_url()>` and pass it as the code parameter. In this case,
|
|
||||||
make sure to also pass the same redirect_uri parameter as you used when
|
|
||||||
generating the auth request URL.
|
|
||||||
|
|
||||||
Returns the access token as a string.
|
|
||||||
"""
|
|
||||||
if username is not None and password is not None:
|
|
||||||
params = self.__generate_params(locals(), ['scopes', 'to_file', 'code', 'refresh_token'])
|
|
||||||
params['grant_type'] = 'password'
|
|
||||||
elif code is not None:
|
|
||||||
params = self.__generate_params(locals(), ['scopes', 'to_file', 'username', 'password', 'refresh_token'])
|
|
||||||
params['grant_type'] = 'authorization_code'
|
|
||||||
elif refresh_token is not None:
|
|
||||||
params = self.__generate_params(locals(), ['scopes', 'to_file', 'username', 'password', 'code'])
|
|
||||||
params['grant_type'] = 'refresh_token'
|
|
||||||
else:
|
|
||||||
raise MastodonIllegalArgumentError('Invalid arguments given. username and password or code are required.')
|
|
||||||
|
|
||||||
params['client_id'] = self.client_id
|
|
||||||
params['client_secret'] = self.client_secret
|
|
||||||
params['scope'] = " ".join(scopes)
|
|
||||||
|
|
||||||
try:
|
|
||||||
response = self.__api_request('POST', '/oauth/token', params, do_ratelimiting=False)
|
|
||||||
self.access_token = response['access_token']
|
|
||||||
self.__set_refresh_token(response.get('refresh_token'))
|
|
||||||
self.__set_token_expired(int(response.get('expires_in', 0)))
|
|
||||||
except Exception as e:
|
|
||||||
if username is not None or password is not None:
|
|
||||||
raise MastodonIllegalArgumentError('Invalid user name, password, or redirect_uris: %s' % e)
|
|
||||||
elif code is not None:
|
|
||||||
raise MastodonIllegalArgumentError('Invalid access token or redirect_uris: %s' % e)
|
|
||||||
else:
|
|
||||||
raise MastodonIllegalArgumentError('Invalid request: %s' % e)
|
|
||||||
|
|
||||||
received_scopes = response["scope"].split(" ")
|
|
||||||
for scope_set in _SCOPE_SETS.keys():
|
|
||||||
if scope_set in received_scopes:
|
|
||||||
received_scopes += _SCOPE_SETS[scope_set]
|
|
||||||
|
|
||||||
if not set(scopes) <= set(received_scopes):
|
|
||||||
raise MastodonAPIError('Granted scopes "' + " ".join(received_scopes) + '" do not contain all of the requested scopes "' + " ".join(scopes) + '".')
|
|
||||||
|
|
||||||
if to_file is not None:
|
|
||||||
with open(to_file, 'w') as token_file:
|
|
||||||
token_file.write(response['access_token'] + "\n")
|
|
||||||
token_file.write(self.api_base_url + "\n")
|
|
||||||
token_file.write(self.client_id + "\n")
|
|
||||||
token_file.write(self.client_secret + "\n")
|
|
||||||
|
|
||||||
self.__logged_in_id = None
|
|
||||||
|
|
||||||
# Retry version check if needed (might be required in limited federation mode)
|
|
||||||
if not self.version_check_worked:
|
|
||||||
self.retrieve_mastodon_version()
|
|
||||||
|
|
||||||
return response['access_token']
|
|
||||||
|
|
||||||
|
|
||||||
def revoke_access_token(self):
|
|
||||||
"""
|
|
||||||
Revoke the oauth token the user is currently authenticated with, effectively removing
|
|
||||||
the apps access and requiring the user to log in again.
|
|
||||||
"""
|
|
||||||
if self.access_token is None:
|
|
||||||
raise MastodonIllegalArgumentError("Not logged in, do not have a token to revoke.")
|
|
||||||
if self.client_id is None or self.client_secret is None:
|
|
||||||
raise MastodonIllegalArgumentError("Client authentication (id + secret) is required to revoke tokens.")
|
|
||||||
params = collections.OrderedDict([])
|
|
||||||
params['client_id'] = self.client_id
|
|
||||||
params['client_secret'] = self.client_secret
|
|
||||||
params['token'] = self.access_token
|
|
||||||
self.__api_request('POST', '/oauth/revoke', params)
|
|
||||||
|
|
||||||
# We are now logged out, clear token and logged in id
|
|
||||||
self.access_token = None
|
|
||||||
self.__logged_in_id = None
|
|
||||||
|
|
||||||
def set_language(self, lang):
|
|
||||||
"""
|
|
||||||
Set the locale Mastodon will use to generate responses. Valid parameters are all ISO 639-1 (two letter) or, for languages that do
|
|
||||||
not have one, 639-3 (three letter) language codes. This affects some error messages (those related to validation) and trends.
|
|
||||||
"""
|
|
||||||
self.lang = lang
|
|
||||||
|
|
||||||
def retrieve_mastodon_version(self):
|
|
||||||
"""
|
|
||||||
Determine installed Mastodon version and set major, minor and patch (not including RC info) accordingly.
|
|
||||||
|
|
||||||
Returns the version string, possibly including rc info.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
version_str = self.__normalize_version_string(self.__instance()["version"])
|
|
||||||
self.version_check_worked = True
|
|
||||||
except:
|
|
||||||
# instance() was added in 1.1.0, so our best guess is 1.0.0.
|
|
||||||
version_str = "1.0.0"
|
|
||||||
self.version_check_worked = False
|
|
||||||
|
|
||||||
self.mastodon_major, self.mastodon_minor, self.mastodon_patch = parse_version_string(version_str)
|
|
||||||
return version_str
|
|
||||||
|
|
||||||
def verify_minimum_version(self, version_str, cached=False):
|
|
||||||
"""
|
|
||||||
Update version info from server and verify that at least the specified version is present.
|
|
||||||
|
|
||||||
If you specify "cached", the version info update part is skipped.
|
|
||||||
|
|
||||||
Returns True if version requirement is satisfied, False if not.
|
|
||||||
"""
|
|
||||||
if not cached:
|
|
||||||
self.retrieve_mastodon_version()
|
|
||||||
major, minor, patch = parse_version_string(version_str)
|
|
||||||
if major > self.mastodon_major:
|
|
||||||
return False
|
|
||||||
elif major == self.mastodon_major and minor > self.mastodon_minor:
|
|
||||||
return False
|
|
||||||
elif major == self.mastodon_major and minor == self.mastodon_minor and patch > self.mastodon_patch:
|
|
||||||
return False
|
|
||||||
return True
|
|
||||||
|
|
||||||
def get_approx_server_time(self):
|
|
||||||
"""
|
|
||||||
Retrieve the approximate server time
|
|
||||||
|
|
||||||
We parse this from the hopefully present "Date" header, but make no effort to compensate for latency.
|
|
||||||
"""
|
|
||||||
response = self.__api_request("HEAD", "/", return_response_object=True)
|
|
||||||
if 'Date' in response.headers:
|
|
||||||
server_time_datetime = dateutil.parser.parse(response.headers['Date'])
|
|
||||||
|
|
||||||
# Make sure we're in local time
|
|
||||||
epoch_time = self.__datetime_to_epoch(server_time_datetime)
|
|
||||||
return datetime.datetime.fromtimestamp(epoch_time)
|
|
||||||
else:
|
|
||||||
raise MastodonAPIError("No server time in response.")
|
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_supported_version():
|
def get_supported_version():
|
||||||
"""
|
"""
|
||||||
@ -479,95 +64,6 @@ class Mastodon(Internals, Accounts):
|
|||||||
"""
|
"""
|
||||||
return Mastodon.__SUPPORTED_MASTODON_VERSION
|
return Mastodon.__SUPPORTED_MASTODON_VERSION
|
||||||
|
|
||||||
###
|
|
||||||
# Reading data: Instances
|
|
||||||
###
|
|
||||||
@api_version("1.1.0", "2.3.0", _DICT_VERSION_INSTANCE)
|
|
||||||
def instance(self):
|
|
||||||
"""
|
|
||||||
Retrieve basic information about the instance, including the URI and administrative contact email.
|
|
||||||
|
|
||||||
Does not require authentication unless locked down by the administrator.
|
|
||||||
|
|
||||||
Returns an :ref:`instance dict <instance dict>`.
|
|
||||||
"""
|
|
||||||
return self.__instance()
|
|
||||||
|
|
||||||
def __instance(self):
|
|
||||||
"""
|
|
||||||
Internal, non-version-checking helper that does the same as instance()
|
|
||||||
"""
|
|
||||||
instance = self.__api_request('GET', '/api/v1/instance/')
|
|
||||||
return instance
|
|
||||||
|
|
||||||
@api_version("2.1.2", "2.1.2", _DICT_VERSION_ACTIVITY)
|
|
||||||
def instance_activity(self):
|
|
||||||
"""
|
|
||||||
Retrieve activity stats about the instance. May be disabled by the instance administrator - throws
|
|
||||||
a MastodonNotFoundError in that case.
|
|
||||||
|
|
||||||
Activity is returned for 12 weeks going back from the current week.
|
|
||||||
|
|
||||||
Returns a list of :ref:`activity dicts <activity dicts>`.
|
|
||||||
"""
|
|
||||||
return self.__api_request('GET', '/api/v1/instance/activity')
|
|
||||||
|
|
||||||
@api_version("2.1.2", "2.1.2", "2.1.2")
|
|
||||||
def instance_peers(self):
|
|
||||||
"""
|
|
||||||
Retrieve the instances that this instance knows about. May be disabled by the instance administrator - throws
|
|
||||||
a MastodonNotFoundError in that case.
|
|
||||||
|
|
||||||
Returns a list of URL strings.
|
|
||||||
"""
|
|
||||||
return self.__api_request('GET', '/api/v1/instance/peers')
|
|
||||||
|
|
||||||
@api_version("3.0.0", "3.0.0", "3.0.0")
|
|
||||||
def instance_health(self):
|
|
||||||
"""
|
|
||||||
Basic health check. Returns True if healthy, False if not.
|
|
||||||
"""
|
|
||||||
status = self.__api_request('GET', '/health', parse=False).decode("utf-8")
|
|
||||||
return status in ["OK", "success"]
|
|
||||||
|
|
||||||
@api_version("3.0.0", "3.0.0", "3.0.0")
|
|
||||||
def instance_nodeinfo(self, schema="http://nodeinfo.diaspora.software/ns/schema/2.0"):
|
|
||||||
"""
|
|
||||||
Retrieves the instance's nodeinfo information.
|
|
||||||
|
|
||||||
For information on what the nodeinfo can contain, see the nodeinfo
|
|
||||||
specification: https://github.com/jhass/nodeinfo . By default,
|
|
||||||
Mastodon.py will try to retrieve the version 2.0 schema nodeinfo.
|
|
||||||
|
|
||||||
To override the schema, specify the desired schema with the `schema`
|
|
||||||
parameter.
|
|
||||||
"""
|
|
||||||
links = self.__api_request('GET', '/.well-known/nodeinfo')["links"]
|
|
||||||
|
|
||||||
schema_url = None
|
|
||||||
for available_schema in links:
|
|
||||||
if available_schema.rel == schema:
|
|
||||||
schema_url = available_schema.href
|
|
||||||
|
|
||||||
if schema_url is None:
|
|
||||||
raise MastodonIllegalArgumentError(
|
|
||||||
"Requested nodeinfo schema is not available.")
|
|
||||||
|
|
||||||
try:
|
|
||||||
return self.__api_request('GET', schema_url, base_url_override="")
|
|
||||||
except MastodonNotFoundError:
|
|
||||||
parse = urlparse(schema_url)
|
|
||||||
return self.__api_request('GET', parse.path + parse.params + parse.query + parse.fragment)
|
|
||||||
|
|
||||||
@api_version("3.4.0", "3.4.0", _DICT_VERSION_INSTANCE)
|
|
||||||
def instance_rules(self):
|
|
||||||
"""
|
|
||||||
Retrieve instance rules.
|
|
||||||
|
|
||||||
Returns a list of `id` + `text` dicts, same as the `rules` field in the :ref:`instance dicts <instance dicts>`.
|
|
||||||
"""
|
|
||||||
return self.__api_request('GET', '/api/v1/instance/rules')
|
|
||||||
|
|
||||||
###
|
###
|
||||||
# Reading data: Timelines
|
# Reading data: Timelines
|
||||||
##
|
##
|
||||||
@ -3379,14 +2875,14 @@ class Mastodon(Internals, Accounts):
|
|||||||
|
|
||||||
push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend())
|
push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend())
|
||||||
push_key_priv = push_key_pair.private_numbers().private_value
|
push_key_priv = push_key_pair.private_numbers().private_value
|
||||||
|
try:
|
||||||
crypto_ver = cryptography.__version__
|
push_key_pub = push_key_pair.public_key().public_bytes(
|
||||||
if len(crypto_ver) < 5:
|
serialization.Encoding.X962,
|
||||||
crypto_ver += ".0"
|
serialization.PublicFormat.UncompressedPoint,
|
||||||
if parse_version_string(crypto_ver) == (2, 5, 0):
|
)
|
||||||
sapush_key_pub = push_key_pair.public_key().public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
|
except:
|
||||||
else:
|
|
||||||
push_key_pub = push_key_pair.public_key().public_numbers().encode_point()
|
push_key_pub = push_key_pair.public_key().public_numbers().encode_point()
|
||||||
|
|
||||||
push_shared_secret = os.urandom(16)
|
push_shared_secret = os.urandom(16)
|
||||||
|
|
||||||
priv_dict = {
|
priv_dict = {
|
||||||
|
|||||||
@ -2,7 +2,9 @@ from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS
|
|||||||
from .error import MastodonIllegalArgumentError, MastodonAPIError
|
from .error import MastodonIllegalArgumentError, MastodonAPIError
|
||||||
from .utility import api_version
|
from .utility import api_version
|
||||||
|
|
||||||
class Mastodon():
|
from .internals import Mastodon as Internals
|
||||||
|
|
||||||
|
class Mastodon(Internals):
|
||||||
@api_version("2.7.0", "2.7.0", "3.4.0")
|
@api_version("2.7.0", "2.7.0", "3.4.0")
|
||||||
def create_account(self, username, password, email, agreement=False, reason=None, locale="en", scopes=_DEFAULT_SCOPES, to_file=None, return_detailed_error=False):
|
def create_account(self, username, password, email, agreement=False, reason=None, locale="en", scopes=_DEFAULT_SCOPES, to_file=None, return_detailed_error=False):
|
||||||
"""
|
"""
|
||||||
|
|||||||
370
mastodon/authentication.py
Normal file
370
mastodon/authentication.py
Normal file
@ -0,0 +1,370 @@
|
|||||||
|
import requests
|
||||||
|
from requests.models import urlencode
|
||||||
|
import datetime
|
||||||
|
import os
|
||||||
|
import time
|
||||||
|
import collections
|
||||||
|
|
||||||
|
from .error import MastodonIllegalArgumentError, MastodonNetworkError, MastodonVersionError, MastodonAPIError
|
||||||
|
from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS, _DEFAULT_TIMEOUT
|
||||||
|
from .utility import parse_version_string
|
||||||
|
from .internals import Mastodon as Internals
|
||||||
|
|
||||||
|
class Mastodon(Internals):
|
||||||
|
###
|
||||||
|
# Registering apps
|
||||||
|
###
|
||||||
|
@staticmethod
|
||||||
|
def create_app(client_name, scopes=_DEFAULT_SCOPES, redirect_uris=None, website=None, to_file=None,
|
||||||
|
api_base_url=None, request_timeout=_DEFAULT_TIMEOUT, session=None):
|
||||||
|
"""
|
||||||
|
Create a new app with given `client_name` and `scopes` (The basic scopes are "read", "write", "follow" and "push"
|
||||||
|
- more granular scopes are available, please refer to Mastodon documentation for which) on the instance given
|
||||||
|
by `api_base_url`.
|
||||||
|
|
||||||
|
Specify `redirect_uris` if you want users to be redirected to a certain page after authenticating in an OAuth flow.
|
||||||
|
You can specify multiple URLs by passing a list. Note that if you wish to use OAuth authentication with redirects,
|
||||||
|
the redirect URI must be one of the URLs specified here.
|
||||||
|
|
||||||
|
Specify `to_file` to persist your app's info to a file so you can use it in the constructor.
|
||||||
|
Specify `website` to give a website for your app.
|
||||||
|
|
||||||
|
Specify `session` with a requests.Session for it to be used instead of the default. This can be
|
||||||
|
used to, amongst other things, adjust proxy or SSL certificate settings.
|
||||||
|
|
||||||
|
Presently, app registration is open by default, but this is not guaranteed to be the case for all
|
||||||
|
Mastodon instances in the future.
|
||||||
|
|
||||||
|
|
||||||
|
Returns `client_id` and `client_secret`, both as strings.
|
||||||
|
"""
|
||||||
|
if api_base_url is None:
|
||||||
|
raise MastodonIllegalArgumentError("API base URL is required.")
|
||||||
|
api_base_url = Mastodon.__protocolize(api_base_url)
|
||||||
|
|
||||||
|
request_data = {
|
||||||
|
'client_name': client_name,
|
||||||
|
'scopes': " ".join(scopes)
|
||||||
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
if redirect_uris is not None:
|
||||||
|
if isinstance(redirect_uris, (list, tuple)):
|
||||||
|
redirect_uris = "\n".join(list(redirect_uris))
|
||||||
|
request_data['redirect_uris'] = redirect_uris
|
||||||
|
else:
|
||||||
|
request_data['redirect_uris'] = 'urn:ietf:wg:oauth:2.0:oob'
|
||||||
|
if website is not None:
|
||||||
|
request_data['website'] = website
|
||||||
|
if session:
|
||||||
|
ret = session.post(api_base_url + '/api/v1/apps', data=request_data, timeout=request_timeout)
|
||||||
|
response = ret.json()
|
||||||
|
else:
|
||||||
|
response = requests.post(api_base_url + '/api/v1/apps', data=request_data, timeout=request_timeout)
|
||||||
|
response = response.json()
|
||||||
|
except Exception as e:
|
||||||
|
raise MastodonNetworkError("Could not complete request: %s" % e)
|
||||||
|
|
||||||
|
if to_file is not None:
|
||||||
|
with open(to_file, 'w') as secret_file:
|
||||||
|
secret_file.write(response['client_id'] + "\n")
|
||||||
|
secret_file.write(response['client_secret'] + "\n")
|
||||||
|
secret_file.write(api_base_url + "\n")
|
||||||
|
secret_file.write(client_name + "\n")
|
||||||
|
|
||||||
|
return (response['client_id'], response['client_secret'])
|
||||||
|
|
||||||
|
###
|
||||||
|
# Authentication, including constructor
|
||||||
|
###
|
||||||
|
def __init__(self, client_id=None, client_secret=None, access_token=None, api_base_url=None, debug_requests=False,
|
||||||
|
ratelimit_method="wait", ratelimit_pacefactor=1.1, request_timeout=_DEFAULT_TIMEOUT, mastodon_version=None,
|
||||||
|
version_check_mode="created", session=None, feature_set="mainline", user_agent="mastodonpy", lang=None):
|
||||||
|
"""
|
||||||
|
Create a new API wrapper instance based on the given `client_secret` and `client_id` on the
|
||||||
|
instance given by `api_base_url`. If you give a `client_id` and it is not a file, you must
|
||||||
|
also give a secret. If you specify an `access_token` then you don't need to specify a `client_id`.
|
||||||
|
It is allowed to specify neither - in this case, you will be restricted to only using endpoints
|
||||||
|
that do not require authentication. If a file is given as `client_id`, client ID, secret and
|
||||||
|
base url are read from that file.
|
||||||
|
|
||||||
|
You can also specify an `access_token`, directly or as a file (as written by :ref:`log_in() <log_in()>`). If
|
||||||
|
a file is given, Mastodon.py also tries to load the base URL from this file, if present. A
|
||||||
|
client id and secret are not required in this case.
|
||||||
|
|
||||||
|
Mastodon.py can try to respect rate limits in several ways, controlled by `ratelimit_method`.
|
||||||
|
"throw" makes functions throw a `MastodonRatelimitError` when the rate
|
||||||
|
limit is hit. "wait" mode will, once the limit is hit, wait and retry the request as soon
|
||||||
|
as the rate limit resets, until it succeeds. "pace" works like throw, but tries to wait in
|
||||||
|
between calls so that the limit is generally not hit (how hard it tries to avoid hitting the rate
|
||||||
|
limit can be controlled by ratelimit_pacefactor). The default setting is "wait". Note that
|
||||||
|
even in "wait" and "pace" mode, requests can still fail due to network or other problems! Also
|
||||||
|
note that "pace" and "wait" are NOT thread safe.
|
||||||
|
|
||||||
|
By default, a timeout of 300 seconds is used for all requests. If you wish to change this,
|
||||||
|
pass the desired timeout (in seconds) as `request_timeout`.
|
||||||
|
|
||||||
|
For fine-tuned control over the requests object use `session` with a requests.Session.
|
||||||
|
|
||||||
|
The `mastodon_version` parameter can be used to specify the version of Mastodon that Mastodon.py will
|
||||||
|
expect to be installed on the server. The function will throw an error if an unparseable
|
||||||
|
Version is specified. If no version is specified, Mastodon.py will set `mastodon_version` to the
|
||||||
|
detected version.
|
||||||
|
|
||||||
|
The version check mode can be set to "created" (the default behaviour), "changed" or "none". If set to
|
||||||
|
"created", Mastodon.py will throw an error if the version of Mastodon it is connected to is too old
|
||||||
|
to have an endpoint. If it is set to "changed", it will throw an error if the endpoint's behaviour has
|
||||||
|
changed after the version of Mastodon that is connected has been released. If it is set to "none",
|
||||||
|
version checking is disabled.
|
||||||
|
|
||||||
|
`feature_set` can be used to enable behaviour specific to non-mainline Mastodon API implementations.
|
||||||
|
Details are documented in the functions that provide such functionality. Currently supported feature
|
||||||
|
sets are `mainline`, `fedibird` and `pleroma`.
|
||||||
|
|
||||||
|
For some Mastodon instances a `User-Agent` header is needed. This can be set by parameter `user_agent`. Starting from
|
||||||
|
Mastodon.py 1.5.2 `create_app()` stores the application name into the client secret file. If `client_id` points to this file,
|
||||||
|
the app name will be used as `User-Agent` header as default. It is possible to modify old secret files and append
|
||||||
|
a client app name to use it as a `User-Agent` name.
|
||||||
|
|
||||||
|
`lang` can be used to change the locale Mastodon will use to generate responses. Valid parameters are all ISO 639-1 (two letter)
|
||||||
|
or for a language that has none, 639-3 (three letter) language codes. This affects some error messages (those related to validation) and
|
||||||
|
trends. You can change the language using :ref:`set_language()`.
|
||||||
|
|
||||||
|
If no other `User-Agent` is specified, "mastodonpy" will be used.
|
||||||
|
"""
|
||||||
|
self.api_base_url = api_base_url
|
||||||
|
if self.api_base_url is not None:
|
||||||
|
self.api_base_url = self.__protocolize(self.api_base_url)
|
||||||
|
self.client_id = client_id
|
||||||
|
self.client_secret = client_secret
|
||||||
|
self.access_token = access_token
|
||||||
|
self.debug_requests = debug_requests
|
||||||
|
self.ratelimit_method = ratelimit_method
|
||||||
|
self._token_expired = datetime.datetime.now()
|
||||||
|
self._refresh_token = None
|
||||||
|
|
||||||
|
self.__logged_in_id = None
|
||||||
|
|
||||||
|
self.ratelimit_limit = 300
|
||||||
|
self.ratelimit_reset = time.time()
|
||||||
|
self.ratelimit_remaining = 300
|
||||||
|
self.ratelimit_lastcall = time.time()
|
||||||
|
self.ratelimit_pacefactor = ratelimit_pacefactor
|
||||||
|
|
||||||
|
self.request_timeout = request_timeout
|
||||||
|
|
||||||
|
if session:
|
||||||
|
self.session = session
|
||||||
|
else:
|
||||||
|
self.session = requests.Session()
|
||||||
|
|
||||||
|
self.feature_set = feature_set
|
||||||
|
if not self.feature_set in ["mainline", "fedibird", "pleroma"]:
|
||||||
|
raise MastodonIllegalArgumentError('Requested invalid feature set')
|
||||||
|
|
||||||
|
# General defined user-agent
|
||||||
|
self.user_agent = user_agent
|
||||||
|
|
||||||
|
# Save language
|
||||||
|
self.lang = lang
|
||||||
|
|
||||||
|
# Token loading
|
||||||
|
if self.client_id is not None:
|
||||||
|
if os.path.isfile(self.client_id):
|
||||||
|
with open(self.client_id, 'r') as secret_file:
|
||||||
|
self.client_id = secret_file.readline().rstrip()
|
||||||
|
self.client_secret = secret_file.readline().rstrip()
|
||||||
|
|
||||||
|
try_base_url = secret_file.readline().rstrip()
|
||||||
|
if try_base_url is not None and len(try_base_url) != 0:
|
||||||
|
try_base_url = Mastodon.__protocolize(try_base_url)
|
||||||
|
if not (self.api_base_url is None or try_base_url == self.api_base_url):
|
||||||
|
raise MastodonIllegalArgumentError('Mismatch in base URLs between files and/or specified')
|
||||||
|
self.api_base_url = try_base_url
|
||||||
|
|
||||||
|
# With new registrations we support the 4th line to store a client_name and use it as user-agent
|
||||||
|
client_name = secret_file.readline()
|
||||||
|
if client_name and self.user_agent is None:
|
||||||
|
self.user_agent = client_name.rstrip()
|
||||||
|
else:
|
||||||
|
if self.client_secret is None:
|
||||||
|
raise MastodonIllegalArgumentError('Specified client id directly, but did not supply secret')
|
||||||
|
|
||||||
|
if self.access_token is not None and os.path.isfile(self.access_token):
|
||||||
|
with open(self.access_token, 'r') as token_file:
|
||||||
|
self.access_token = token_file.readline().rstrip()
|
||||||
|
|
||||||
|
# For newer versions, we also store the URL
|
||||||
|
try_base_url = token_file.readline().rstrip()
|
||||||
|
if try_base_url is not None and len(try_base_url) != 0:
|
||||||
|
try_base_url = Mastodon.__protocolize(try_base_url)
|
||||||
|
if not (self.api_base_url is None or try_base_url == self.api_base_url):
|
||||||
|
raise MastodonIllegalArgumentError('Mismatch in base URLs between files and/or specified')
|
||||||
|
self.api_base_url = try_base_url
|
||||||
|
|
||||||
|
# For EVEN newer vesions, we ALSO ALSO store the client id and secret so that you don't need to reauth to revoke
|
||||||
|
if self.client_id is None:
|
||||||
|
try:
|
||||||
|
self.client_id = token_file.readline().rstrip()
|
||||||
|
self.client_secret = token_file.readline().rstrip()
|
||||||
|
except:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Verify we have a base URL, protocolize
|
||||||
|
if self.api_base_url is None:
|
||||||
|
raise MastodonIllegalArgumentError("API base URL is required.")
|
||||||
|
self.api_base_url = Mastodon.__protocolize(self.api_base_url)
|
||||||
|
|
||||||
|
if not version_check_mode in ["created", "changed", "none"]:
|
||||||
|
raise MastodonIllegalArgumentError("Invalid version check method.")
|
||||||
|
self.version_check_mode = version_check_mode
|
||||||
|
|
||||||
|
self.mastodon_major = 1
|
||||||
|
self.mastodon_minor = 0
|
||||||
|
self.mastodon_patch = 0
|
||||||
|
self.version_check_worked = None
|
||||||
|
|
||||||
|
# Versioning
|
||||||
|
if mastodon_version is None and self.version_check_mode != 'none':
|
||||||
|
self.retrieve_mastodon_version()
|
||||||
|
elif self.version_check_mode != 'none':
|
||||||
|
try:
|
||||||
|
self.mastodon_major, self.mastodon_minor, self.mastodon_patch = parse_version_string(mastodon_version)
|
||||||
|
except:
|
||||||
|
raise MastodonVersionError("Bad version specified")
|
||||||
|
|
||||||
|
# Ratelimiting parameter check
|
||||||
|
if ratelimit_method not in ["throw", "wait", "pace"]:
|
||||||
|
raise MastodonIllegalArgumentError("Invalid ratelimit method.")
|
||||||
|
|
||||||
|
def auth_request_url(self, client_id=None, redirect_uris="urn:ietf:wg:oauth:2.0:oob", scopes=_DEFAULT_SCOPES, force_login=False, state=None, lang=None):
|
||||||
|
"""
|
||||||
|
Returns the URL that a client needs to request an OAuth grant from the server.
|
||||||
|
|
||||||
|
To log in with OAuth, send your user to this URL. The user will then log in and
|
||||||
|
get a code which you can pass to :ref:`log_in() <log_in()>`.
|
||||||
|
|
||||||
|
`scopes` are as in :ref:`log_in() <log_in()>`, redirect_uris is where the user should be redirected to
|
||||||
|
after authentication. Note that `redirect_uris` must be one of the URLs given during
|
||||||
|
app registration. When using urn:ietf:wg:oauth:2.0:oob, the code is simply displayed,
|
||||||
|
otherwise it is added to the given URL as the "code" request parameter.
|
||||||
|
|
||||||
|
Pass force_login if you want the user to always log in even when already logged
|
||||||
|
into web Mastodon (i.e. when registering multiple different accounts in an app).
|
||||||
|
|
||||||
|
`state` is the oauth `state` parameter to pass to the server. It is strongly suggested
|
||||||
|
to use a random, nonguessable value (i.e. nothing meaningful and no incrementing ID)
|
||||||
|
to preserve security guarantees. It can be left out for non-web login flows.
|
||||||
|
|
||||||
|
Pass an ISO 639-1 (two letter) or, for languages that do not have one, 639-3 (three letter)
|
||||||
|
language code as `lang` to control the display language for the oauth form.
|
||||||
|
"""
|
||||||
|
if client_id is None:
|
||||||
|
client_id = self.client_id
|
||||||
|
else:
|
||||||
|
if os.path.isfile(client_id):
|
||||||
|
with open(client_id, 'r') as secret_file:
|
||||||
|
client_id = secret_file.readline().rstrip()
|
||||||
|
|
||||||
|
params = dict()
|
||||||
|
params['client_id'] = client_id
|
||||||
|
params['response_type'] = "code"
|
||||||
|
params['redirect_uri'] = redirect_uris
|
||||||
|
params['scope'] = " ".join(scopes)
|
||||||
|
params['force_login'] = force_login
|
||||||
|
params['state'] = state
|
||||||
|
params['lang'] = lang
|
||||||
|
formatted_params = urlencode(params)
|
||||||
|
return "".join([self.api_base_url, "/oauth/authorize?", formatted_params])
|
||||||
|
|
||||||
|
def log_in(self, username=None, password=None, code=None, redirect_uri="urn:ietf:wg:oauth:2.0:oob", refresh_token=None, scopes=_DEFAULT_SCOPES, to_file=None):
|
||||||
|
"""
|
||||||
|
Get the access token for a user.
|
||||||
|
|
||||||
|
The username is the email address used to log in into Mastodon.
|
||||||
|
|
||||||
|
Can persist access token to file `to_file`, to be used in the constructor.
|
||||||
|
|
||||||
|
Handles password and OAuth-based authorization.
|
||||||
|
|
||||||
|
Will throw a `MastodonIllegalArgumentError` if the OAuth flow data or the
|
||||||
|
username / password credentials given are incorrect, and
|
||||||
|
`MastodonAPIError` if all of the requested scopes were not granted.
|
||||||
|
|
||||||
|
For OAuth 2, obtain a code via having your user go to the URL returned by
|
||||||
|
:ref:`auth_request_url() <auth_request_url()>` and pass it as the code parameter. In this case,
|
||||||
|
make sure to also pass the same redirect_uri parameter as you used when
|
||||||
|
generating the auth request URL.
|
||||||
|
|
||||||
|
Returns the access token as a string.
|
||||||
|
"""
|
||||||
|
if username is not None and password is not None:
|
||||||
|
params = self.__generate_params(locals(), ['scopes', 'to_file', 'code', 'refresh_token'])
|
||||||
|
params['grant_type'] = 'password'
|
||||||
|
elif code is not None:
|
||||||
|
params = self.__generate_params(locals(), ['scopes', 'to_file', 'username', 'password', 'refresh_token'])
|
||||||
|
params['grant_type'] = 'authorization_code'
|
||||||
|
elif refresh_token is not None:
|
||||||
|
params = self.__generate_params(locals(), ['scopes', 'to_file', 'username', 'password', 'code'])
|
||||||
|
params['grant_type'] = 'refresh_token'
|
||||||
|
else:
|
||||||
|
raise MastodonIllegalArgumentError('Invalid arguments given. username and password or code are required.')
|
||||||
|
|
||||||
|
params['client_id'] = self.client_id
|
||||||
|
params['client_secret'] = self.client_secret
|
||||||
|
params['scope'] = " ".join(scopes)
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = self.__api_request('POST', '/oauth/token', params, do_ratelimiting=False)
|
||||||
|
self.access_token = response['access_token']
|
||||||
|
self.__set_refresh_token(response.get('refresh_token'))
|
||||||
|
self.__set_token_expired(int(response.get('expires_in', 0)))
|
||||||
|
except Exception as e:
|
||||||
|
if username is not None or password is not None:
|
||||||
|
raise MastodonIllegalArgumentError('Invalid user name, password, or redirect_uris: %s' % e)
|
||||||
|
elif code is not None:
|
||||||
|
raise MastodonIllegalArgumentError('Invalid access token or redirect_uris: %s' % e)
|
||||||
|
else:
|
||||||
|
raise MastodonIllegalArgumentError('Invalid request: %s' % e)
|
||||||
|
|
||||||
|
received_scopes = response["scope"].split(" ")
|
||||||
|
for scope_set in _SCOPE_SETS.keys():
|
||||||
|
if scope_set in received_scopes:
|
||||||
|
received_scopes += _SCOPE_SETS[scope_set]
|
||||||
|
|
||||||
|
if not set(scopes) <= set(received_scopes):
|
||||||
|
raise MastodonAPIError('Granted scopes "' + " ".join(received_scopes) + '" do not contain all of the requested scopes "' + " ".join(scopes) + '".')
|
||||||
|
|
||||||
|
if to_file is not None:
|
||||||
|
with open(to_file, 'w') as token_file:
|
||||||
|
token_file.write(response['access_token'] + "\n")
|
||||||
|
token_file.write(self.api_base_url + "\n")
|
||||||
|
token_file.write(self.client_id + "\n")
|
||||||
|
token_file.write(self.client_secret + "\n")
|
||||||
|
|
||||||
|
self.__logged_in_id = None
|
||||||
|
|
||||||
|
# Retry version check if needed (might be required in limited federation mode)
|
||||||
|
if not self.version_check_worked:
|
||||||
|
self.retrieve_mastodon_version()
|
||||||
|
|
||||||
|
return response['access_token']
|
||||||
|
|
||||||
|
def revoke_access_token(self):
|
||||||
|
"""
|
||||||
|
Revoke the oauth token the user is currently authenticated with, effectively removing
|
||||||
|
the apps access and requiring the user to log in again.
|
||||||
|
"""
|
||||||
|
if self.access_token is None:
|
||||||
|
raise MastodonIllegalArgumentError("Not logged in, do not have a token to revoke.")
|
||||||
|
if self.client_id is None or self.client_secret is None:
|
||||||
|
raise MastodonIllegalArgumentError("Client authentication (id + secret) is required to revoke tokens.")
|
||||||
|
params = collections.OrderedDict([])
|
||||||
|
params['client_id'] = self.client_id
|
||||||
|
params['client_secret'] = self.client_secret
|
||||||
|
params['token'] = self.access_token
|
||||||
|
self.__api_request('POST', '/oauth/revoke', params)
|
||||||
|
|
||||||
|
# We are now logged out, clear token and logged in id
|
||||||
|
self.access_token = None
|
||||||
|
self.__logged_in_id = None
|
||||||
96
mastodon/instance.py
Normal file
96
mastodon/instance.py
Normal file
@ -0,0 +1,96 @@
|
|||||||
|
from .versions import _DICT_VERSION_INSTANCE, _DICT_VERSION_ACTIVITY
|
||||||
|
from .error import MastodonIllegalArgumentError, MastodonNotFoundError
|
||||||
|
from .utility import api_version
|
||||||
|
from .compat import urlparse
|
||||||
|
|
||||||
|
from .internals import Mastodon as Internals
|
||||||
|
|
||||||
|
class Mastodon(Internals):
|
||||||
|
###
|
||||||
|
# Reading data: Instances
|
||||||
|
###
|
||||||
|
@api_version("1.1.0", "2.3.0", _DICT_VERSION_INSTANCE)
|
||||||
|
def instance(self):
|
||||||
|
"""
|
||||||
|
Retrieve basic information about the instance, including the URI and administrative contact email.
|
||||||
|
|
||||||
|
Does not require authentication unless locked down by the administrator.
|
||||||
|
|
||||||
|
Returns an :ref:`instance dict <instance dict>`.
|
||||||
|
"""
|
||||||
|
return self.__instance()
|
||||||
|
|
||||||
|
def __instance(self):
|
||||||
|
"""
|
||||||
|
Internal, non-version-checking helper that does the same as instance()
|
||||||
|
"""
|
||||||
|
instance = self.__api_request('GET', '/api/v1/instance/')
|
||||||
|
return instance
|
||||||
|
|
||||||
|
@api_version("2.1.2", "2.1.2", _DICT_VERSION_ACTIVITY)
|
||||||
|
def instance_activity(self):
|
||||||
|
"""
|
||||||
|
Retrieve activity stats about the instance. May be disabled by the instance administrator - throws
|
||||||
|
a MastodonNotFoundError in that case.
|
||||||
|
|
||||||
|
Activity is returned for 12 weeks going back from the current week.
|
||||||
|
|
||||||
|
Returns a list of :ref:`activity dicts <activity dicts>`.
|
||||||
|
"""
|
||||||
|
return self.__api_request('GET', '/api/v1/instance/activity')
|
||||||
|
|
||||||
|
@api_version("2.1.2", "2.1.2", "2.1.2")
|
||||||
|
def instance_peers(self):
|
||||||
|
"""
|
||||||
|
Retrieve the instances that this instance knows about. May be disabled by the instance administrator - throws
|
||||||
|
a MastodonNotFoundError in that case.
|
||||||
|
|
||||||
|
Returns a list of URL strings.
|
||||||
|
"""
|
||||||
|
return self.__api_request('GET', '/api/v1/instance/peers')
|
||||||
|
|
||||||
|
@api_version("3.0.0", "3.0.0", "3.0.0")
|
||||||
|
def instance_health(self):
|
||||||
|
"""
|
||||||
|
Basic health check. Returns True if healthy, False if not.
|
||||||
|
"""
|
||||||
|
status = self.__api_request('GET', '/health', parse=False).decode("utf-8")
|
||||||
|
return status in ["OK", "success"]
|
||||||
|
|
||||||
|
@api_version("3.0.0", "3.0.0", "3.0.0")
|
||||||
|
def instance_nodeinfo(self, schema="http://nodeinfo.diaspora.software/ns/schema/2.0"):
|
||||||
|
"""
|
||||||
|
Retrieves the instance's nodeinfo information.
|
||||||
|
|
||||||
|
For information on what the nodeinfo can contain, see the nodeinfo
|
||||||
|
specification: https://github.com/jhass/nodeinfo . By default,
|
||||||
|
Mastodon.py will try to retrieve the version 2.0 schema nodeinfo.
|
||||||
|
|
||||||
|
To override the schema, specify the desired schema with the `schema`
|
||||||
|
parameter.
|
||||||
|
"""
|
||||||
|
links = self.__api_request('GET', '/.well-known/nodeinfo')["links"]
|
||||||
|
|
||||||
|
schema_url = None
|
||||||
|
for available_schema in links:
|
||||||
|
if available_schema.rel == schema:
|
||||||
|
schema_url = available_schema.href
|
||||||
|
|
||||||
|
if schema_url is None:
|
||||||
|
raise MastodonIllegalArgumentError(
|
||||||
|
"Requested nodeinfo schema is not available.")
|
||||||
|
|
||||||
|
try:
|
||||||
|
return self.__api_request('GET', schema_url, base_url_override="")
|
||||||
|
except MastodonNotFoundError:
|
||||||
|
parse = urlparse(schema_url)
|
||||||
|
return self.__api_request('GET', parse.path + parse.params + parse.query + parse.fragment)
|
||||||
|
|
||||||
|
@api_version("3.4.0", "3.4.0", _DICT_VERSION_INSTANCE)
|
||||||
|
def instance_rules(self):
|
||||||
|
"""
|
||||||
|
Retrieve instance rules.
|
||||||
|
|
||||||
|
Returns a list of `id` + `text` dicts, same as the `rules` field in the :ref:`instance dicts <instance dicts>`.
|
||||||
|
"""
|
||||||
|
return self.__api_request('GET', '/api/v1/instance/rules')
|
||||||
@ -13,7 +13,7 @@ import collections
|
|||||||
import base64
|
import base64
|
||||||
import os
|
import os
|
||||||
|
|
||||||
from .utility import AttribAccessDict, AttribAccessList
|
from .utility import AttribAccessDict, AttribAccessList, parse_version_string
|
||||||
from .error import MastodonNetworkError, MastodonIllegalArgumentError, MastodonRatelimitError, MastodonNotFoundError, \
|
from .error import MastodonNetworkError, MastodonIllegalArgumentError, MastodonRatelimitError, MastodonNotFoundError, \
|
||||||
MastodonUnauthorizedError, MastodonInternalServerError, MastodonBadGatewayError, MastodonServiceUnavailableError, \
|
MastodonUnauthorizedError, MastodonInternalServerError, MastodonBadGatewayError, MastodonServiceUnavailableError, \
|
||||||
MastodonGatewayTimeoutError, MastodonServerError, MastodonAPIError, MastodonMalformedEventError
|
MastodonGatewayTimeoutError, MastodonServerError, MastodonAPIError, MastodonMalformedEventError
|
||||||
|
|||||||
@ -2,7 +2,11 @@
|
|||||||
|
|
||||||
import re
|
import re
|
||||||
from decorator import decorate
|
from decorator import decorate
|
||||||
from .error import MastodonVersionError
|
from .error import MastodonVersionError, MastodonAPIError
|
||||||
|
import dateutil
|
||||||
|
import datetime
|
||||||
|
|
||||||
|
# Module level:
|
||||||
|
|
||||||
###
|
###
|
||||||
# Version check functions, including decorator and parser
|
# Version check functions, including decorator and parser
|
||||||
@ -74,4 +78,66 @@ class AttribAccessList(list):
|
|||||||
def __setattr__(self, attr, val):
|
def __setattr__(self, attr, val):
|
||||||
if attr in self:
|
if attr in self:
|
||||||
raise AttributeError("Attribute-style access is read only")
|
raise AttributeError("Attribute-style access is read only")
|
||||||
super(AttribAccessList, self).__setattr__(attr, val)
|
super(AttribAccessList, self).__setattr__(attr, val)
|
||||||
|
|
||||||
|
|
||||||
|
# Class level:
|
||||||
|
class Mastodon():
|
||||||
|
def set_language(self, lang):
|
||||||
|
"""
|
||||||
|
Set the locale Mastodon will use to generate responses. Valid parameters are all ISO 639-1 (two letter) or, for languages that do
|
||||||
|
not have one, 639-3 (three letter) language codes. This affects some error messages (those related to validation) and trends.
|
||||||
|
"""
|
||||||
|
self.lang = lang
|
||||||
|
|
||||||
|
def retrieve_mastodon_version(self):
|
||||||
|
"""
|
||||||
|
Determine installed Mastodon version and set major, minor and patch (not including RC info) accordingly.
|
||||||
|
|
||||||
|
Returns the version string, possibly including rc info.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
version_str = self.__normalize_version_string(self.__instance()["version"])
|
||||||
|
self.version_check_worked = True
|
||||||
|
except:
|
||||||
|
# instance() was added in 1.1.0, so our best guess is 1.0.0.
|
||||||
|
version_str = "1.0.0"
|
||||||
|
self.version_check_worked = False
|
||||||
|
|
||||||
|
self.mastodon_major, self.mastodon_minor, self.mastodon_patch = parse_version_string(version_str)
|
||||||
|
return version_str
|
||||||
|
|
||||||
|
def verify_minimum_version(self, version_str, cached=False):
|
||||||
|
"""
|
||||||
|
Update version info from server and verify that at least the specified version is present.
|
||||||
|
|
||||||
|
If you specify "cached", the version info update part is skipped.
|
||||||
|
|
||||||
|
Returns True if version requirement is satisfied, False if not.
|
||||||
|
"""
|
||||||
|
if not cached:
|
||||||
|
self.retrieve_mastodon_version()
|
||||||
|
major, minor, patch = parse_version_string(version_str)
|
||||||
|
if major > self.mastodon_major:
|
||||||
|
return False
|
||||||
|
elif major == self.mastodon_major and minor > self.mastodon_minor:
|
||||||
|
return False
|
||||||
|
elif major == self.mastodon_major and minor == self.mastodon_minor and patch > self.mastodon_patch:
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
def get_approx_server_time(self):
|
||||||
|
"""
|
||||||
|
Retrieve the approximate server time
|
||||||
|
|
||||||
|
We parse this from the hopefully present "Date" header, but make no effort to compensate for latency.
|
||||||
|
"""
|
||||||
|
response = self.__api_request("HEAD", "/", return_response_object=True)
|
||||||
|
if 'Date' in response.headers:
|
||||||
|
server_time_datetime = dateutil.parser.parse(response.headers['Date'])
|
||||||
|
|
||||||
|
# Make sure we're in local time
|
||||||
|
epoch_time = self.__datetime_to_epoch(server_time_datetime)
|
||||||
|
return datetime.datetime.fromtimestamp(epoch_time)
|
||||||
|
else:
|
||||||
|
raise MastodonAPIError("No server time in response.")
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user