This commit is contained in:
2026-04-08 18:14:11 +09:00
parent c4968a6b78
commit 8caa642dda
7 changed files with 41 additions and 16 deletions

View File

@@ -1,7 +1,7 @@
{% extends "/base.html" %} {% extends "/base.html" %}
{% block title %}{{ status_code }} {{ status_code_name }} - Nercone{% endblock %} {% block title %}{{ status_code }} {{ status_code_name }} - Nercone{% endblock %}
{% block extra_title %}Error{% endblock %} {% block extra_title %}{{ status_code }}{% endblock %}
{% block header_desc %}{{ status_code }} {{ status_code_name }}{% endblock %} {% block header_desc %}<span class="text-lowlight">{{ joke_message }}</span>{% endblock %}
{% block content %} {% block content %}
<h1 class="text-bold">{{ status_code }} {{ status_code_name }}</h1> <h1 class="text-bold">{{ status_code }} {{ status_code_name }}</h1>
<p>{{ message }}</p> <p>{{ message }}</p>

View File

@@ -1,6 +1,7 @@
import sqlite3 import sqlite3
from pathlib import Path from pathlib import Path
# TODO: アクセスカウンターのためだけにDB必要無くね && アクセスログをDB化して行をカウントするだけで良くね && 構造的にDB設計するの面倒そうだからこのままでええかも
class AccessCounter: class AccessCounter:
def __init__(self, filepath: str = str(Path.cwd().joinpath("databases", "access_counter.db"))): def __init__(self, filepath: str = str(Path.cwd().joinpath("databases", "access_counter.db"))):
self.filepath = filepath self.filepath = filepath

View File

@@ -2,6 +2,7 @@ from http import HTTPStatus
from fastapi import Request, Response from fastapi import Request, Response
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
# TODO: こっちは普通のメッセージにする
default_messages = { default_messages = {
400: "日本語でおk", 400: "日本語でおk",
401: "見たいのならログインすることね", 401: "見たいのならログインすることね",
@@ -19,11 +20,37 @@ default_messages = {
413: "そ、そそ、そんなの入りきらないよっ!", 413: "そ、そそ、そんなの入りきらないよっ!",
414: "もちつけ", 414: "もちつけ",
415: "そんな形式知らない!", 415: "そんな形式知らない!",
416: "そんな大きく...ない...んだ...", 416: "ちっさぁ:heart:",
417: "期待させて悪かったわね!", 417: "期待させて悪かったわね!",
418: "ティーポット「私はコーヒーを注ぐためのものではありません!やだっ!」" 418: "ティーポット「私はコーヒーを注ぐためのものではありません!やだっ!」",
421: "またあいつ案内先間違えてるよ...どうしよ...",
426: "それに答えるには、まずWebSocketに移動したい。"
} }
def error_page(templates: Jinja2Templates, request: Request, status_code: int, message: str | None = None) -> Response: default_joke_messages = {
400: "日本語でおk",
401: "見たいのならログインすることね",
402: "夢が欲しけりゃ金払え!",
403: "あんたなんかに見せるもんですか!",
404: "そんなページ知らないっ!",
405: "そのMethodはNot Allowedだよ",
406: "すまんがその条件ではお渡しできない。",
407: "うちのプロキシ使うんだったらまずログインしな。",
408: "もう用がないならさっさと帰りなさい。",
409: "ちょっと待ったそんな話聞いてないぞ",
410: "もう無いで。",
411: "サイズを教えろ。話はそれからだ。",
412: "なにその条件美味しいの",
413: "そ、そそ、そんなの入りきらないよっ!",
414: "もちつけ",
415: "そんな形式知らない!",
416: "ちっさぁ:heart:",
417: "期待させて悪かったわね!",
418: "ティーポット「私はコーヒーを注ぐためのものではありません!やだっ!」",
421: "またあいつ案内先間違えてるよ...どうしよ...",
426: "それに答えるには、まずWebSocketに移動したい。"
}
def error_page(templates: Jinja2Templates, request: Request, status_code: int, message: str | None = None, joke_message: str | None = None) -> Response:
status_code_name = HTTPStatus(status_code).phrase status_code_name = HTTPStatus(status_code).phrase
return templates.TemplateResponse(status_code=status_code, request=request, name="error.html", context={"status_code": status_code, "status_code_name": status_code_name, "message": message or default_messages.get(status_code, "あんのーん")}) return templates.TemplateResponse(status_code=status_code, request=request, name="error.html", context={"status_code": status_code, "status_code_name": status_code_name, "message": message or default_messages.get(status_code, "不明なエラーが発生してしまったようです。ご迷惑をおかけし申し訳ございません..."), "joke_message": joke_message or default_joke_messages.get(status_code, "あんのーん")})

View File

@@ -7,6 +7,7 @@ from datetime import datetime, timezone
access_log_path = Path.cwd().joinpath("logs", "access.log") access_log_path = Path.cwd().joinpath("logs", "access.log")
access_log_path.parent.mkdir(parents=True, exist_ok=True) access_log_path.parent.mkdir(parents=True, exist_ok=True)
# TODO: ステータスコードとかも含めたい でもそれにはレスポンスが出来上がってからログを書く必要があるので/echoとかを残すには工夫がいりそう
def log_access(scope: Scope, write: bool = True): def log_access(scope: Scope, write: bool = True):
client = scope.get("client") or ("", 0) client = scope.get("client") or ("", 0)
server = scope.get("server") or ("", 0) server = scope.get("server") or ("", 0)

View File

@@ -1,5 +1,4 @@
import subprocess import subprocess
from datetime import datetime, timezone
from fastapi import Response from fastapi import Response
from fastapi.responses import PlainTextResponse from fastapi.responses import PlainTextResponse
from starlette.types import Scope, ASGIApp, Receive, Send from starlette.types import Scope, ASGIApp, Receive, Send

View File

@@ -1,4 +1,3 @@
import re
import httpx import httpx
import asyncio import asyncio
from websockets.client import connect from websockets.client import connect
@@ -6,20 +5,18 @@ from fastapi import Request, Response, WebSocket
hop_by_hop_headers = ["transfer-encoding", "connection", "keep-alive", "upgrade", "proxy-authenticate", "proxy-authorization", "te", "trailers", "content-encoding", "content-length"] hop_by_hop_headers = ["transfer-encoding", "connection", "keep-alive", "upgrade", "proxy-authenticate", "proxy-authorization", "te", "trailers", "content-encoding", "content-length"]
def to_raw_headers(str_headers: dict[str, str]) -> list[tuple[bytes, bytes]]:
return [(k.encode("latin-1"), v.encode("latin-1")) for k, v in str_headers.items()]
def make_http_proxy(base_url_http: str, headers: dict = {}, remove_prefix_path: bool = False): def make_http_proxy(base_url_http: str, headers: dict = {}, remove_prefix_path: bool = False):
async def http_proxy(request: Request, path: str = "") -> Response: async def http_proxy(request: Request, path: str = "") -> Response:
url = f"{base_url_http}/{path}" if remove_prefix_path else f"{base_url_http}{request.url.path}" url = f"{base_url_http}/{path}" if remove_prefix_path else f"{base_url_http}{request.url.path}"
merged_headers = dict(request.headers) merged_headers = dict(request.headers)
merged_headers.pop("accept-encoding", None) merged_headers.pop("accept-encoding", None)
merged_headers |= {k.lower(): v for k, v in headers.items()} merged_headers |= {k.lower(): v for k, v in headers.items()}
raw_headers = [(k.encode("latin-1"), v.encode("latin-1")) for k, v in merged_headers.items()]
async with httpx.AsyncClient() as client: async with httpx.AsyncClient() as client:
resp = await client.request( resp = await client.request(
method=request.method, method=request.method,
url=url, url=url,
headers=to_raw_headers(merged_headers), headers=raw_headers,
content=await request.body(), content=await request.body(),
params=request.query_params params=request.query_params
) )

View File

@@ -99,7 +99,7 @@ async def default_response(request: Request, full_path: str) -> Response:
if static := resolve_static_file(full_path): if static := resolve_static_file(full_path):
return FileResponse(static) return FileResponse(static)
except PermissionError: except PermissionError:
return error_page(templates, request, 403, "ディレクトリトラバーサルね、知ってる。公開してないところ覗きたいのえっt") return error_page(templates, request, 403, "何をしてるんです?脆弱性報告のためならいいのですが、データ盗んで悪用するためなら今すぐにやめてくださいね?", "ディレクトリトラバーサルね、知ってる。公開してないところ覗きたいのえっt")
if full_path in ["", "/"]: if full_path in ["", "/"]:
template_candidates = ["index.html"] template_candidates = ["index.html"]
@@ -127,12 +127,12 @@ async def default_response(request: Request, full_path: str) -> Response:
try: try:
path = Path.cwd().joinpath("public", "shorturls.json") path = Path.cwd().joinpath("public", "shorturls.json")
if not path.exists(): if not path.exists():
return error_page(templates, request, 500, "設定ファイルぐらい用意しておけよ!") return error_page(templates, request, 500, "短縮URLの処理のためのJSONファイルがありません。", "設定ファイルぐらい用意しておけよ!")
shorturls = json.load(path.open("r", encoding="utf-8")) shorturls = json.load(path.open("r", encoding="utf-8"))
except Exception: except Exception:
return error_page(templates, request, 500, "なにこの設定ファイル読めないじゃない!") return error_page(templates, request, 500, "短縮URLの処理のためのJSONファイルを正常に読み込めませんでした。", "なにこの設定ファイル読めないじゃない!")
if result := resolve_shorturl(shorturls, full_path): if result := resolve_shorturl(shorturls, full_path):
return RedirectResponse(url=result) return RedirectResponse(url=result)
return error_page(templates, request, 404, "そんなページ知らないっ!") return error_page(templates, request, 404, "リクエストしたページは現在ご利用になれません。削除/移動されたか、URLが間違っている可能性があります。", "そんなページ知らないっ!")