From 5c210dc20f4c3dd12472f44e31228fa1f4a96b79 Mon Sep 17 00:00:00 2001 From: Izalia Mae Date: Sat, 16 Mar 2024 01:36:20 -0400 Subject: [PATCH] fix linting issues --- relay/application.py | 34 +++++++++++++++++----------------- relay/database/connection.py | 2 +- relay/dev.py | 10 +++++----- relay/views/api.py | 7 ++++--- relay/views/base.py | 2 +- relay/views/frontend.py | 12 +++++------- 6 files changed, 33 insertions(+), 34 deletions(-) diff --git a/relay/application.py b/relay/application.py index cb32357..abe8258 100644 --- a/relay/application.py +++ b/relay/application.py @@ -35,6 +35,21 @@ if typing.TYPE_CHECKING: from .misc import Message, Response +def get_csp(request: web.Request) -> str: + data = [ + "default-src 'none'", + f"script-src 'nonce-{request['hash']}'", + f"style-src 'self' 'nonce-{request['hash']}'", + "form-action 'self'", + "connect-src 'self'", + "img-src 'self'", + "object-src 'none'", + "frame-ancestors 'none'" + ] + + return '; '.join(data) + ';' + + class Application(web.Application): DEFAULT: Application | None = None @@ -127,21 +142,6 @@ class Application(web.Application): return timedelta(seconds=uptime.seconds) - def get_csp(self, request: Request) -> str: - data = [ - "default-src 'none'", - f"script-src 'nonce-{request['hash']}'", - f"style-src 'self' 'nonce-{request['hash']}'", - "form-action 'self'", - "connect-src 'self'", - "img-src 'self'", - "object-src 'none'", - "frame-ancestors 'none'" - ] - - return '; '.join(data) + ';' - - def push_message(self, inbox: str, message: Message, instance: Row) -> None: self['push_queue'].put((inbox, message, instance)) @@ -240,7 +240,7 @@ class CachedStaticResource(StaticResource): def __init__(self, prefix: str, path: Path): StaticResource.__init__(self, prefix, path) - self.cache: dict[Path, bytes] = {} + self.cache: dict[str, bytes] = {} for filename in path.rglob('*'): if filename.is_dir(): @@ -334,7 +334,7 @@ async def handle_response_headers(request: web.Request, handler: Callable) -> Re # Still have to figure out how csp headers work if resp.content_type == 'text/html': - resp.headers['Content-Security-Policy'] = Application.DEFAULT.get_csp(request) + resp.headers['Content-Security-Policy'] = get_csp(request) if not request.app['dev'] and request.path.endswith(('.css', '.js')): # cache for 2 weeks diff --git a/relay/database/connection.py b/relay/database/connection.py index 4f9ff21..67df706 100644 --- a/relay/database/connection.py +++ b/relay/database/connection.py @@ -192,7 +192,7 @@ class Connection(SqlConnection): def put_user(self, username: str, password: str | None, handle: str | None = None) -> Row: if self.get_user(username): - data = { + data: dict[str, str | datetime | None] = { 'username': username } diff --git a/relay/dev.py b/relay/dev.py index f145d24..c03becd 100644 --- a/relay/dev.py +++ b/relay/dev.py @@ -100,8 +100,8 @@ def cli_run(dev: bool): try: while True: - handler.proc.stdin.write(sys.stdin.read().encode('UTF-8')) - handler.proc.stdin.flush() + handler.proc.stdin.write(sys.stdin.read().encode('UTF-8')) # type: ignore + handler.proc.stdin.flush() # type: ignore except KeyboardInterrupt: pass @@ -121,12 +121,12 @@ class WatchHandler(PatternMatchingEventHandler): PatternMatchingEventHandler.__init__(self) self.dev: bool = dev - self.proc = None - self.last_restart = None + self.proc: subprocess.Popen | None = None + self.last_restart: datetime | None = None def kill_proc(self): - if self.proc.poll() is not None: + if not self.proc or self.proc.poll() is not None: return logging.info(f'Terminating process {self.proc.pid}') diff --git a/relay/views/api.py b/relay/views/api.py index 1789f72..92e29d8 100644 --- a/relay/views/api.py +++ b/relay/views/api.py @@ -15,6 +15,7 @@ from ..misc import Message, Response, boolean, get_app if typing.TYPE_CHECKING: from aiohttp.web import Request from collections.abc import Callable, Sequence + from typing import Any PUBLIC_API_PATHS: Sequence[tuple[str, str]] = ( @@ -149,7 +150,7 @@ class Config(View): if isinstance(data, Response): return data - data['key'] = data['key'].replace('-', '_'); + data['key'] = data['key'].replace('-', '_') if data['key'] not in ConfigData.USER_KEYS(): return Response.new_error(400, 'Invalid key', 'json') @@ -255,7 +256,7 @@ class RequestView(View): async def post(self, request: Request) -> Response: - data = await self.get_api_data(['domain', 'accept'], []) + data: dict[str, Any] | Response = await self.get_api_data(['domain', 'accept'], []) data['accept'] = boolean(data['accept']) try: @@ -430,7 +431,7 @@ class User(View): async def patch(self, request: Request) -> Response: - data = await self.get_api_data(['username'], ['password', ['handle']]) + data = await self.get_api_data(['username'], ['password', 'handle']) if isinstance(data, Response): return data diff --git a/relay/views/base.py b/relay/views/base.py index 9709e34..3d7d718 100644 --- a/relay/views/base.py +++ b/relay/views/base.py @@ -126,7 +126,7 @@ class View(AbstractView): return Response.new_error(400, 'Invalid JSON data', 'json') else: - post_data = convert_data(self.request.query) # type: ignore + post_data = convert_data(self.request.query) data = {} diff --git a/relay/views/frontend.py b/relay/views/frontend.py index 206482b..91af960 100644 --- a/relay/views/frontend.py +++ b/relay/views/frontend.py @@ -3,14 +3,12 @@ from __future__ import annotations import typing from aiohttp import web -from argon2.exceptions import VerifyMismatchError -from urllib.parse import urlparse from .base import View, register_route -from ..database import THEMES, ConfigData +from ..database import THEMES from ..logger import LogLevel -from ..misc import ACTOR_FORMATS, Message, Response, get_app +from ..misc import Response, get_app if typing.TYPE_CHECKING: from aiohttp.web import Request @@ -40,9 +38,9 @@ async def handle_frontend_path(request: web.Request, handler: Callable) -> Respo return Response.new('', 302, {'Location': '/'}) if not request['user'] and request.path.startswith('/admin'): - response = Response.new('', 302, {'Location': f'/login?redir={request.path}'}) - response.del_cookie('user-token') - return response + response = Response.new('', 302, {'Location': f'/login?redir={request.path}'}) + response.del_cookie('user-token') + return response response = await handler(request)