iartamonov/mautrix-telegram
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Improve type hints and set version to 0.4.0+dev

Browse Source
This commit is contained in:
Tulir Asokan
2018-09-10 01:14:12 +03:00
parent 4b2cdc3d39
commit d4ea5f8b38
21 changed files with 200 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mautrix_telegram/web/common/auth_api.py
+18 -12
View File
@@ -15,6 +15,9 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from abc import abstractmethod
from typing import Optional
from aiohttp import web
import abc
import asyncio
import logging
@@ -28,22 +31,24 @@ from ...user import User
class AuthAPI(abc.ABC):
log = logging.getLogger("mau.web.auth")
log = logging.getLogger("mau.web.auth") # type: logging.Logger
def __init__(self, loop):
def __init__(self, loop: asyncio.AbstractEventLoop):
self.loop = loop # type: asyncio.AbstractEventLoop
@abstractmethod
def get_login_response(self, status=200, state="", username="", mxid="", message="", error="",
errcode=""):
def get_login_response(self, status: int = 200, state: str = "", username: str = "",
mxid: str = "", message: str = "", error: str = "",
errcode: str = "") -> web.Response:
raise NotImplementedError()
@abstractmethod
def get_mx_login_response(self, status=200, state="", username="", mxid="", message="",
error="", errcode=""):
def get_mx_login_response(self, status: int = 200, state: str = "", username: str = "",
mxid: str = "", message: str = "", error: str = "",
errcode: str = "") -> web.Response:
raise NotImplementedError()
async def post_matrix_token(self, user: User, token):
async def post_matrix_token(self, user: User, token: str) -> web.Response:
puppet = Puppet.get(user.tgid)
if puppet.is_real_user:
return self.get_mx_login_response(state="already-logged-in", status=409,
@@ -61,11 +66,11 @@ class AuthAPI(abc.ABC):
return self.get_mx_login_response(mxid=user.mxid, status=200, state="logged-in")
async def post_matrix_password(self, user, password):
async def post_matrix_password(self, user: User, password: str) -> web.Response:
return self.get_mx_login_response(mxid=user.mxid, status=501, error="Not yet implemented",
errcode="not-yet-implemented")
async def post_login_phone(self, user, phone):
async def post_login_phone(self, user: User, phone: str) -> web.Response:
try:
await user.client.sign_in(phone or "+123")
return self.get_login_response(mxid=user.mxid, state="code", status=200,
@@ -102,7 +107,7 @@ class AuthAPI(abc.ABC):
errcode="unknown_error",
error="Internal server error while requesting code.")
async def post_login_token(self, user, token):
async def post_login_token(self, user: User, token: str) -> web.Response:
try:
user_info = await user.client.sign_in(bot_token=token)
asyncio.ensure_future(user.post_login(user_info), loop=self.loop)
@@ -123,7 +128,8 @@ class AuthAPI(abc.ABC):
return self.get_login_response(mxid=user.mxid, state="token", status=500,
error="Internal server error while sending token.")
async def post_login_code(self, user, code, password_in_data):
async def post_login_code(self, user: User, code: int, password_in_data: bool
) -> Optional[web.Response]:
try:
user_info = await user.client.sign_in(code=code)
asyncio.ensure_future(user.post_login(user_info), loop=self.loop)
@@ -156,7 +162,7 @@ class AuthAPI(abc.ABC):
errcode="unknown_error",
error="Internal server error while sending code.")
async def post_login_password(self, user, password):
async def post_login_password(self, user: User, password: str) -> web.Response:
try:
user_info = await user.client.sign_in(password=password)
asyncio.ensure_future(user.post_login(user_info), loop=self.loop)
mautrix_telegram/web/provisioning/__init__.py
+6 -4
View File
@@ -35,15 +35,16 @@ if TYPE_CHECKING:
class ProvisioningAPI(AuthAPI):
log = logging.getLogger("mau.web.provisioning")
log = logging.getLogger("mau.web.provisioning") # type: logging.Logger
def __init__(self, context: "Context") -> None:
super().__init__(context.loop)
self.secret = context.config["appservice.provisioning.shared_secret"]
self.secret = context.config["appservice.provisioning.shared_secret"] # type: str
self.az = context.az # type: AppService
self.context = context # type: Context
self.app = web.Application(loop=context.loop, middlewares=[self.error_middleware])
self.app = web.Application(loop=context.loop, middlewares=[self.error_middleware]
) # type: web.Application
portal_prefix = "/portal/{mxid:![^/]+}"
self.app.router.add_route("GET", f"{portal_prefix}", self.get_portal_by_mxid)
@@ -353,7 +354,8 @@ class ProvisioningAPI(AuthAPI):
await user.log_out()
@staticmethod
async def error_middleware(_, handler) -> Callable[[web.Request], Awaitable[web.Response]]:
async def error_middleware(_, handler: Callable[[web.Request], Awaitable[web.Response]]
) -> Callable[[web.Request], Awaitable[web.Response]]:
async def middleware_handler(request: web.Request) -> web.Response:
try:
return await handler(request)
mautrix_telegram/web/public/__init__.py
+24 -19
View File
@@ -14,14 +14,17 @@
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from typing import Optional
from aiohttp import web
from mako.template import Template
import pkg_resources
import asyncio
import logging
import random
import string
import time
from ...types import MatrixUserID
from ...util import sign_token, verify_token
from ...user import User
from ...puppet import Puppet
@@ -29,20 +32,20 @@ from ..common import AuthAPI
class PublicBridgeWebsite(AuthAPI):
log = logging.getLogger("mau.web.public")
log = logging.getLogger("mau.web.public") # type: logging.Logger
def __init__(self, loop):
def __init__(self, loop: asyncio.AbstractEventLoop):
super().__init__(loop)
self.secret_key = "".join(
random.choice(string.ascii_lowercase + string.digits) for _ in range(64))
random.choice(string.ascii_lowercase + string.digits) for _ in range(64)) # type: str
self.login = Template(
pkg_resources.resource_string("mautrix_telegram", "web/public/login.html.mako"))
self.login = Template(pkg_resources.resource_string(
"mautrix_telegram", "web/public/login.html.mako")) # type: Template
self.mx_login = Template(
pkg_resources.resource_string("mautrix_telegram", "web/public/matrix-login.html.mako"))
self.mx_login = Template(pkg_resources.resource_string(
"mautrix_telegram", "web/public/matrix-login.html.mako")) # type: Template
self.app = web.Application(loop=loop)
self.app = web.Application(loop=loop) # type: web.Application
self.app.router.add_route("GET", "/login", self.get_login)
self.app.router.add_route("POST", "/login", self.post_login)
self.app.router.add_route("GET", "/matrix-login", self.get_matrix_login)
@@ -50,21 +53,21 @@ class PublicBridgeWebsite(AuthAPI):
self.app.router.add_static("/", pkg_resources.resource_filename("mautrix_telegram",
"web/public/"))
def make_token(self, mxid, endpoint="/login", expires_in=900):
def make_token(self, mxid: str, endpoint: str = "/login", expires_in: int = 900) -> str:
return sign_token(self.secret_key, {
"mxid": mxid,
"endpoint": endpoint,
"expiry": int(time.time()) + expires_in,
})
def verify_token(self, token, endpoint="/login"):
def verify_token(self, token: str, endpoint: str = "/login") -> Optional[MatrixUserID]:
token = verify_token(self.secret_key, token)
if token and (token.get("expiry", 0) > int(time.time()) and
token.get("endpoint", None) == endpoint):
return token.get("mxid", None)
return MatrixUserID(token.get("mxid", None))
return None
async def get_login(self, request):
async def get_login(self, request: web.Request) -> web.Response:
state = "bot_token" if request.rel_url.query.get("mode", "") == "bot" else "request"
mxid = self.verify_token(request.rel_url.query.get("token", None), endpoint="/login")
@@ -83,7 +86,7 @@ class PublicBridgeWebsite(AuthAPI):
return self.get_login_response(mxid=user.mxid, username=user.username)
async def get_matrix_login(self, request):
async def get_matrix_login(self, request: web.Request) -> web.Response:
mxid = self.verify_token(request.rel_url.query.get("token", None), endpoint="/matrix-login")
if not mxid:
return self.get_mx_login_response(status=401, state="invalid-token")
@@ -105,19 +108,21 @@ class PublicBridgeWebsite(AuthAPI):
return self.get_mx_login_response(mxid=user.mxid)
def get_login_response(self, status=200, state="", username="", mxid="", message="", error="",
errcode=""):
def get_login_response(self, status: int = 200, state: str = "", username: str = "",
mxid: str = "", message: str = "", error: str = "",
errcode: str = "") -> web.Response:
return web.Response(status=status, content_type="text/html",
text=self.login.render(username=username, state=state, error=error,
message=message, mxid=mxid))
def get_mx_login_response(self, status=200, state="", username="", mxid="", message="",
error="", errcode=""):
def get_mx_login_response(self, status: int = 200, state: str = "", username: str = "",
mxid: str = "", message: str = "", error: str = "",
errcode: str = "") -> web.Response:
return web.Response(status=status, content_type="text/html",
text=self.mx_login.render(username=username, state=state, error=error,
message=message, mxid=mxid))
async def post_matrix_login(self, request):
async def post_matrix_login(self, request: web.Request) -> web.Response:
mxid = self.verify_token(request.rel_url.query.get("token", None), endpoint="/matrix-login")
if not mxid:
return self.get_mx_login_response(status=401, state="invalid-token")
@@ -140,7 +145,7 @@ class PublicBridgeWebsite(AuthAPI):
error="You must provide an access token or "
"password.")
async def post_login(self, request):
async def post_login(self, request: web.Request) -> web.Response:
mxid = self.verify_token(request.rel_url.query.get("token", None), endpoint="/login")
if not mxid:
return self.get_login_response(status=401, state="invalid-token")