From d9cd7547fd0919a46b79ede084910b7df66f8fd7 Mon Sep 17 00:00:00 2001 From: halcy Date: Wed, 30 Nov 2022 23:47:46 +0200 Subject: Move some more methods, fix crypto heckup --- mastodon/Mastodon.py | 526 +-------------------------------------------- mastodon/accounts.py | 4 +- mastodon/authentication.py | 370 +++++++++++++++++++++++++++++++ mastodon/instance.py | 96 +++++++++ mastodon/internals.py | 2 +- mastodon/utility.py | 70 +++++- 6 files changed, 549 insertions(+), 519 deletions(-) create mode 100644 mastodon/authentication.py create mode 100644 mastodon/instance.py (limited to 'mastodon') diff --git a/mastodon/Mastodon.py b/mastodon/Mastodon.py index 76dccd3..35b8444 100644 --- a/mastodon/Mastodon.py +++ b/mastodon/Mastodon.py @@ -24,6 +24,7 @@ from .compat import urlparse from .utility import parse_version_string, max_version, api_version from .utility import AttribAccessDict, AttribAccessDict +from .utility import Mastodon as Utility from .error import * 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 .internals import Mastodon as Internals +from .authentication import Mastodon as Authentication from .accounts import Mastodon as Accounts +from .instance import Mastodon as Instance ## # The actual Mastodon class ### -class Mastodon(Internals, Accounts): +class Mastodon(Utility, Authentication, Accounts, Instance): """ Thorough and easy to use Mastodon API wrapper in Python. @@ -54,424 +57,6 @@ class Mastodon(Internals, Accounts): # Support level __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() `). 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() `. - - `scopes` are as in :ref:`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() ` 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 def get_supported_version(): """ @@ -479,95 +64,6 @@ class Mastodon(Internals, Accounts): """ 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 `. - """ - 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 `. - """ - 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 `. - """ - return self.__api_request('GET', '/api/v1/instance/rules') - ### # Reading data: Timelines ## @@ -3379,14 +2875,14 @@ class Mastodon(Internals, Accounts): push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend()) push_key_priv = push_key_pair.private_numbers().private_value - - crypto_ver = cryptography.__version__ - if len(crypto_ver) < 5: - crypto_ver += ".0" - 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) - else: + try: + push_key_pub = push_key_pair.public_key().public_bytes( + serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint, + ) + except: push_key_pub = push_key_pair.public_key().public_numbers().encode_point() + push_shared_secret = os.urandom(16) priv_dict = { diff --git a/mastodon/accounts.py b/mastodon/accounts.py index dcdd8de..5ecdf93 100644 --- a/mastodon/accounts.py +++ b/mastodon/accounts.py @@ -2,7 +2,9 @@ from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS from .error import MastodonIllegalArgumentError, MastodonAPIError 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") def create_account(self, username, password, email, agreement=False, reason=None, locale="en", scopes=_DEFAULT_SCOPES, to_file=None, return_detailed_error=False): """ diff --git a/mastodon/authentication.py b/mastodon/authentication.py new file mode 100644 index 0000000..b7f15c9 --- /dev/null +++ b/mastodon/authentication.py @@ -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() `). 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() `. + + `scopes` are as in :ref:`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() ` 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 \ No newline at end of file diff --git a/mastodon/instance.py b/mastodon/instance.py new file mode 100644 index 0000000..88445d1 --- /dev/null +++ b/mastodon/instance.py @@ -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 `. + """ + 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 `. + """ + 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 `. + """ + return self.__api_request('GET', '/api/v1/instance/rules') diff --git a/mastodon/internals.py b/mastodon/internals.py index 4ee2c5b..0e77421 100644 --- a/mastodon/internals.py +++ b/mastodon/internals.py @@ -13,7 +13,7 @@ import collections import base64 import os -from .utility import AttribAccessDict, AttribAccessList +from .utility import AttribAccessDict, AttribAccessList, parse_version_string from .error import MastodonNetworkError, MastodonIllegalArgumentError, MastodonRatelimitError, MastodonNotFoundError, \ MastodonUnauthorizedError, MastodonInternalServerError, MastodonBadGatewayError, MastodonServiceUnavailableError, \ MastodonGatewayTimeoutError, MastodonServerError, MastodonAPIError, MastodonMalformedEventError diff --git a/mastodon/utility.py b/mastodon/utility.py index f393aa8..53980b6 100644 --- a/mastodon/utility.py +++ b/mastodon/utility.py @@ -2,7 +2,11 @@ import re 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 @@ -74,4 +78,66 @@ class AttribAccessList(list): def __setattr__(self, attr, val): if attr in self: raise AttributeError("Attribute-style access is read only") - super(AttribAccessList, self).__setattr__(attr, val) \ No newline at end of file + 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.") -- cgit v1.2.3