75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
|
from loguru import logger
|
||
|
|
from mastodon.config import MastodonConfiguration
|
||
|
|
import requests
|
||
|
|
import urllib.parse
|
||
|
|
|
||
|
|
class MastodonAuthorization():
|
||
|
|
|
||
|
|
def __init__(self, configuration):
|
||
|
|
self.config = configuration
|
||
|
|
self.scopes = 'read write follow'
|
||
|
|
|
||
|
|
self.register_application()
|
||
|
|
if self.config.get_value("authorization_code") == "":
|
||
|
|
print("Authorize using this link: {}".format(self.get_login_url()))
|
||
|
|
auth_code = input("Paste auth code here:")
|
||
|
|
self.config.set_value("authorization_code", auth_code)
|
||
|
|
self.config.save()
|
||
|
|
|
||
|
|
def authorize(self) -> None:
|
||
|
|
if self.config.get_value("authorization_code") == "":
|
||
|
|
print("Authorize using this link: {}".format(self.get_login_url()))
|
||
|
|
auth_code = input("Paste auth code here:")
|
||
|
|
self.config.set_value("authorization_code", auth_code)
|
||
|
|
self.config.save()
|
||
|
|
|
||
|
|
def fetch_token(self) -> str:
|
||
|
|
if self.config.get_value("access_token") == "":
|
||
|
|
url = "https://"+self.config.get_value("server")+"/oauth/token"
|
||
|
|
|
||
|
|
response = requests.post(url, params={
|
||
|
|
"client_id": self.config.get_value("client_id"),
|
||
|
|
"client_secret":self.config.get_value("client_secret"),
|
||
|
|
"redirect_uri": "urn:ietf:wg:oauth:2.0:oob",
|
||
|
|
"grant_type": "authorization_code",
|
||
|
|
"code":self.config.get_value("authorization_code")
|
||
|
|
})
|
||
|
|
|
||
|
|
response_json = response.json()
|
||
|
|
if response.status_code == 200:
|
||
|
|
self.config.set_value("access_token", response_json["access_token"])
|
||
|
|
self.config.save()
|
||
|
|
else:
|
||
|
|
logger.error("Could not obtain token. Response Code: {}, Error: {}".format(response.status_code, response_json["error"]))
|
||
|
|
return self.config.get_value("access_token")
|
||
|
|
|
||
|
|
def get_auth_headers(self):
|
||
|
|
return {"Authorization": "Bearer {}".format(self.fetch_token())}
|
||
|
|
|
||
|
|
def get_login_url(self):
|
||
|
|
"""Returns the URL for manual log in via browser"""
|
||
|
|
return "https://"+self.config.get_value("server")+"/oauth/authorize/?{}".format(urllib.parse.urlencode({
|
||
|
|
"response_type": "code",
|
||
|
|
"redirect_uri": "urn:ietf:wg:oauth:2.0:oob",
|
||
|
|
"scope": self.scopes,
|
||
|
|
"client_id": self.config.get_value("client_id"),
|
||
|
|
}))
|
||
|
|
|
||
|
|
def register_application(self):
|
||
|
|
logger.debug("Register Application Called")
|
||
|
|
if self.config.get_value("client_secret") == "":
|
||
|
|
logger.debug("Client Secret was empty")
|
||
|
|
url = "https://"+self.config.get_value("server")+"/api/v1/apps"
|
||
|
|
json = {
|
||
|
|
'client_name': self.config.application_name,
|
||
|
|
'redirect_uris': 'urn:ietf:wg:oauth:2.0:oob',
|
||
|
|
'scopes': self.scopes,
|
||
|
|
'website': "",
|
||
|
|
}
|
||
|
|
|
||
|
|
response = requests.post(url, data=json).json()
|
||
|
|
logger.debug(response)
|
||
|
|
self.config.set_value("client_id", response["client_id"])
|
||
|
|
self.config.set_value("client_secret", response["client_secret"])
|
||
|
|
self.config.save()
|