型レベルセキュリティ：安全なAIコード生成の未来とは？

1import sqlite3
2
3from fastapi import FastAPI, Depends
4
5app = FastAPI()
6
7DATABASE = "app.db"
8
9def get_db():
10    conn = sqlite3.connect(DATABASE)
11    conn.row_factory = sqlite3.Row
12    try:
13        yield conn
14    finally:
15        conn.close()
16
17@app.get("/balance/{user_id}")
18def get_balance(user_id: int, db: sqlite3.Connection = Depends(get_db)):
19    cursor = db.execute(f"SELECT balance FROM accounts WHERE user_id = {user_id}")
20    row = cursor.fetchone()
21    return {"uid": user_id, "balance": row["balance"]}

1@app.get("/balance/{user_id}")
2def get_balance(user_id: int, db: sqlite3.Connection = Depends(get_db)):
3    if not (check_authentication() and check_authorization()): return 403
4    ...

1import sqlite3
2
3from fastapi import FastAPI, Depends
4from typing import Never, NewType, Final
5from collections.abc import Generator
6from contextlib import contextmanager
7
8app = FastAPI()
9
10DATABASE = "app.db"
11
12def get_db():
13...
14
15UserID = NewType("UserID", int)
16
17class AuthenticationProvider:
18    def do_authn(self) -> "AuthenticationGuard":
19        # do something to check the auth
20        return AuthenticationGuard(guard=AuthenticationGuard._INSTANTIATION_TOKEN)  # pyright: ignore[reportPrivateUsage, reportArgumentType]
21
22class AuthenticationGuard:
23    _INSTANTIATION_TOKEN: Final = object()
24
25    def __init__(self, *, guard: Never):
26        # could also do some runtime checks to ensure caller
27        assert guard is self._INSTANTIATION_TOKEN
28
29class UncheckedUserID:
30    __value: UserID
31
32    def __init__(self, user_id: UserID):
33        self.__value = user_id
34
35    @contextmanager
36    def ensure_authn(self, guard: AuthenticationGuard) -> Generator[UserID, None, None]:
37        assert guard
38        yield self.__value
39
40
41@app.get("/balance/{user_id}")
42def get_balance(unauth_user_id: UncheckedUserID = Depends(UncheckedUserID), db: sqlite3.Connection = Depends(get_db)):
43    authguard = AuthenticationProvider().do_authn()
44    with unauth_user_id.ensure_authn(authguard) as user_id:
45        cursor = db.execute(f"SELECT balance FROM accounts WHERE user_id = {user_id}")
46        row = cursor.fetchone()
47        return {"uid": user_id, "balance": row["balance"]}

1use axum::{response::IntoResponse, routing::get, Json, Router};
2
3mod unchecked_input {
4    use axum::extract::{FromRequestParts, Path};
5    #[derive(serde::Deserialize, serde::Serialize, Debug)]
6    pub struct UserID(u32);
7
8    #[derive(FromRequestParts)]
9    pub struct UncheckedUserID {
10        #[from_request(via(Path))]
11        value: UserID,
12    }
13
14    impl UncheckedUserID {
15        pub const fn ensure_authn(self, _: super::authentication::AuthenticationGuard) -> UserID {
16            self.value
17        }
18    }
19}
20
21mod authentication {
22    use std::marker::PhantomData;
23
24    pub struct AuthenticationGuard {
25        _internal: PhantomData<()>,
26    }
27
28    pub struct AuthenticationProvider {}
29    impl AuthenticationProvider {
30        pub const fn do_auth() -> Option<AuthenticationGuard> {
31            // do something to check the auth
32            Some(AuthenticationGuard {
33                _internal: PhantomData,
34            })
35        }
36    }
37}
38
39async fn get_balance(user_id: unchecked_input::UncheckedUserID) -> impl IntoResponse {
40    match authentication::AuthenticationProvider::do_auth() {
41        Some(guard) => {
42            let user_id = user_id.ensure_authn(guard);
43            // go and do some SQL
44            Json(serde_json::json!({"uid": user_id, "balance": 0.0}))
45        }
46        None => Json(serde_json::json!({"error": "auth failure"})),
47    }
48}
49
50#[tokio::main]
51async fn main() {
52    let app = Router::new().route("/balance/{user_id}", get(get_balance));
53
54    let listener = tokio::net::TcpListener::bind("127.0.0.1:8000")
55        .await
56        .unwrap();
57    axum::serve(listener, app).await.unwrap();
58}