SantaSpeen/GitflicAPI
1
0
Fork 0
mirror of https://gitflic.ru/project/santaspeen/gitflic.git synced 2025-07-02 02:05:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

oauth

Browse Source
This commit is contained in:
SantaSpeen 2022-03-28 19:08:39 +03:00
parent 68478dd22f
commit 666a320345
2 changed files with 31 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
gitflic/_oauth_server/GitflicOAuthServer.py
View File

@ -1,7 +1,7 @@
import http.server import http.server
import json
import os import os
import sys import sys
import time
from typing import Tuple, AnyStr, NoReturn from typing import Tuple, AnyStr, NoReturn
from urllib.parse import parse_qs, urlsplit from urllib.parse import parse_qs, urlsplit
import importlib.resources import importlib.resources
@ -81,7 +81,6 @@ class GitflicOAuthServer(http.server.BaseHTTPRequestHandler):
<hr/> <hr/>
<center> <center>
{self.server_version.replace("/", " ")} by <a href="https://github.com/SantaSpeen">SantaSpeen</a> {self.server_version.replace("/", " ")} by <a href="https://github.com/SantaSpeen">SantaSpeen</a>
</center>''') </center>''')
else: else:
for k, v in params.items(): for k, v in params.items():
@ -105,14 +104,14 @@ class GitflicOAuthServer(http.server.BaseHTTPRequestHandler):
res = requests.get("https://oauth.gitflic.ru/api/token/access?code=" + code) res = requests.get("https://oauth.gitflic.ru/api/token/access?code=" + code)
if res.status_code == 200: if res.status_code == 200:
res = res.json() jsn = {"request": {"code": code, "state": state}, "response": res.json()}
with open(os.path.join(os.getcwd(), "config.json"), "w") as f: with open(os.path.join(os.getcwd(), "config.json"), "w") as f:
f.write(res) json.dump(jsn, f, indent=3)
access_token = res['accessToken'] access_token = jsn['response']['accessToken']
if _GitflicAuth: if _GitflicAuth:
_GitflicAuth.access_token = access_token _GitflicAuth.access_token = access_token
_GitflicAuth.refresh_token = res['refreshToken'] _GitflicAuth.refresh_token = jsn['response']['refreshToken']
self.send_html_row("access_token", "Access Token:", access_token) self.send_html_row("access_token", "Access Token:", access_token)
else: else:
is_error = True is_error = True
@ -132,7 +131,8 @@ class GitflicOAuthServer(http.server.BaseHTTPRequestHandler):
# Info and footer send # Info and footer send
self.send(""" self.send("""
<div class="text-center bg-light mt-auto" style="position: absolute;width: 100%;background-color: rgba(0, 0, 0, 0.2);bottom: 0;"> <div style="min-height: 53vh !important;"> </div>
<div class="text-center bg-light mt-auto container-fluid">
<div class="text-muted lh-1"> <div class="text-muted lh-1">
<p class="fs-2">Server close automatically.</p> <p class="fs-2">Server close automatically.</p>
<p class="fs-">You can close this page</p> <p class="fs-">You can close this page</p>
@ -160,20 +160,26 @@ class GitflicOAuthServer(http.server.BaseHTTPRequestHandler):
sys.exit(1) sys.exit(1)
def get_server(auth_class, port: int = None) -> Tuple[http.server.HTTPServer, str]: def get_server(auth_class, port: int = None) -> Tuple[http.server.HTTPServer, str, str]:
global server, redirect_link, _GitflicAuth global server, redirect_link, _GitflicAuth
_GitflicAuth = auth_class _GitflicAuth = auth_class
url = "https://gitflic.santaspeen.ru/"
try: try:
port = port or randint(49152, 65535) port = port or randint(49152, 65535)
server_address = ('127.0.0.1', port) server_address = ('localhost', port)
server = http.server.HTTPServer(server_address, GitflicOAuthServer) req = requests.get(url+"getstate", params={"port": port})
redirect_link = f"http://{':'.join([str(i) for i in server_address])}/" if req.status_code == 201:
print(f"GitflicOAuthServer serving on {redirect_link}") server = http.server.HTTPServer(server_address, GitflicOAuthServer)
return server, redirect_link serving = f"http://{':'.join([str(i) for i in server_address])}/"
print(f"GitflicOAuthServer serving on {serving}")
state = req.json()['state']
return server, url+state, state
except Exception as e: except Exception as e:
print(f"get_server send error: {e}. Trying to rebind server.") print(f"get_server send error: {e}. Trying to rebind server.")
return get_server(auth_class) return get_server(auth_class)
raise Exception(f"Cannot get state code from {url}")
def kill_server(): def kill_server():
exit(0) exit(0)

31
gitflic/auth.py
View File

@ -1,6 +1,7 @@
""" """
Gitflic authentication wrapper. Gitflic authentication wrapper.
""" """
import json
import os import os
import threading import threading
from urllib.parse import quote_plus, parse_qs, urlsplit from urllib.parse import quote_plus, parse_qs, urlsplit
@ -64,8 +65,7 @@ class GitflicAuth:
scope: Union[GitflicAuthScopes, str] = GitflicAuthScopes.ALL_READ, scope: Union[GitflicAuthScopes, str] = GitflicAuthScopes.ALL_READ,
client_id: str = "cc2a5d8a-385a-4367-8b2b-bb2412eacb73", client_id: str = "cc2a5d8a-385a-4367-8b2b-bb2412eacb73",
redirect_url: str = "https://gitflic.ru/settings/oauth/token", redirect_url: str = "https://gitflic.ru/settings/oauth/token",
state: str = "python_user", state: str = None):
oauth_implemented_pass: bool = False):
""" """
:param access_token: Raw token for raw AUTH. :param access_token: Raw token for raw AUTH.
:param scope: OAUTH field. Default GitflicAuthScopes.ALL_READ :param scope: OAUTH field. Default GitflicAuthScopes.ALL_READ
@ -100,7 +100,6 @@ class GitflicAuth:
self.state: str = state self.state: str = state
self._server_thread: threading.Thread = None self._server_thread: threading.Thread = None
self._oauth_implemented_pass: bool = oauth_implemented_pass
self._try_login() self._try_login()
@ -140,34 +139,26 @@ class GitflicAuth:
self.log.debug("Trying to login with OAUTH...") self.log.debug("Trying to login with OAUTH...")
if not self._oauth_implemented_pass:
raise GitflicExceptions("OAUTH not implemented yet! Use raw access_token authorization.")
del self._oauth_implemented_pass
if self._localhost_oauth: if self._localhost_oauth:
server, self.redirect_url = GitflicOAuthServer.get_server(self) server, self.redirect_url, self.state = GitflicOAuthServer.get_server(self)
# OAUTH authorization. # OAUTH authorization.
redirect_url = quote_plus(self.redirect_url) redirect_url = quote_plus(self.redirect_url)
webbrowser.open(OAUTH_URL.format(self.scope, self.client_id, redirect_url, self.state)) webbrowser.open(OAUTH_URL.format(self.scope, self.client_id, redirect_url, self.state))
if not self._localhost_oauth: if not self._localhost_oauth:
url_or_code = input("Paste redirect url: ") code = input("Paste code: ")
params = parse_qs(urlsplit(url_or_code).query) if not code:
code = None raise AuthError("Cannot find code.")
for k, v in params.items(): res = self.session.get("https://oauth.gitflic.ru/api/token/access?code=" + code)
if k == "code":
code = v[0]
res = self.session.get("https://oauth.gitflic.ru/api/token/access?code=" + code or url_or_code)
if res.status_code == 200: if res.status_code == 200:
res = res.json() res = res.json()
jsn = {"request": {"code": code, "state": None}, "response": res}
with open(os.path.join(os.getcwd(), "config.json"), "w") as f: with open(os.path.join(os.getcwd(), "config.json"), "w") as f:
f.write(res) json.dump(jsn, f, indent=3)
access_token = res['accessToken'] access_token = res['accessToken']
self.access_token = access_token self.access_token = access_token
self.refresh_token = res['refreshToken'] self.refresh_token = res['refreshToken']
self.session.headers["Authorization"] += access_token self.session.headers["Authorization"] += access_token
self.check_token()
else: else:
error = None error = None
title_split = res.json()['title'].split(".") title_split = res.json()['title'].split(".")
@ -179,7 +170,9 @@ class GitflicAuth:
self._server_thread.start() self._server_thread.start()
print("Waiting server..") print("Waiting server..")
self._server_thread.join() self._server_thread.join()
self.check_token()
self.session.headers['Authorization'] += self.access_token
self.check_token()
def _token_login(self): def _token_login(self):
""" """