模块 14: 软件、安全与综合实践

目录需列出搜索图书并借阅可用副本。不变式为可用副本加已提交借阅等于总数。Catalogue 管 SQL 与事务，Handler 管 HTTP、身份与响应，运行器管启动清理。明确职责帮助定位故障。教学服务单线程且负载有界，公网部署还需生产服务器、TLS、运维监控与完整身份生命周期。

JSON 可解析仍是不可信输入。修改状态前检查结构、精确字段、类型、范围与大小。bool 是 int 子类，因此严格 type 检查排除 True 图书 ID。拒绝未知字段，不静默接受 admin。限制请求长度和读取超时，先规范化再验证语义。这些检查执行约定，不决定谁有权限。

认证识别调用者，授权决定操作权限。项目把随机 bearer 凭据映射到会员与权限范围。缺失或错误凭据为 401，有效只读身份借阅为 403。会员来自认证身份而非 JSON。持有 bearer 即可使用，应避免日志记录并在本地实验外保护传输。若使用密码，应采用带盐专用密码哈希，而非普通快速哈希。

借阅具有请求 ID。在 BEGIN IMMEDIATE 中先检查是否已提交。相同图书会员重放返回已有结果，不再次扣减；同 ID 不同输入冲突。新请求条件扣正库存并同事务插借阅。这样处理模块 11 的超时歧义。幂等记录需持久化，不可仅用重启丢失缓存。项目分别测试正确与不匹配重放。

单元检查隔离验证，集成检查让真实 HTTP 穿过身份、解析、SQL 与响应。测试成功、缺凭据、权限不足、类型错、注入形数据、重复 ID、无库存及重开持久性，再用 SQL 独立检查不变式。默认启动临时服务检查后关闭，--serve 使用指定文件持续至中断。这些证据不等于生产负载或多进程并发测试。

下载脚本，在终端中使用 Python 3.11 或更新版本运行：python m14_validation.py. Windows 也可使用 py -3；部分系统使用 python3。实验仅用标准库。先预测结果，再运行并完成变体。不要使用 -O，以保留断言。下方输出由构建器实际运行捕获，两种语言使用相同代码与输出。

验证器返回规范数据，不修改输入，拒绝空书名、布尔数量、负数和额外字段。

下载 m14_validation.py

"""Boundary validation, normalisation and mutation-free failure cases."""
def validate_book(record):
    if not isinstance(record, dict) or set(record) != {"title", "copies"}:
        raise ValueError("exactly title and copies required")
    if not isinstance(record["title"], str): raise ValueError("text title required")
    title = record["title"].strip()
    if not 1 <= len(title) <= 100: raise ValueError("title length 1..100 required")
    if type(record["copies"]) is not int or not 0 <= record["copies"] <= 100:
        raise ValueError("copies must be an integer in 0..100")
    return {"title": title, "copies": record["copies"]}

good = {"title": "  Dune  ", "copies": 2}
print("normalised:", validate_book(good))
assert good["title"] == "  Dune  "
bad = [{"title": " ", "copies": 1}, {"title": "Dune", "copies": True},
       {"title": "Dune", "copies": -1}, {"title": "Dune", "copies": 1, "admin": True}]
for record in bad:
    try: validate_book(record)
    except ValueError: print("rejected:", repr(record))
    else: raise AssertionError("invalid record accepted")

normalised: {'title': 'Dune', 'copies': 2}
rejected: {'title': ' ', 'copies': 1}
rejected: {'title': 'Dune', 'copies': True}
rejected: {'title': 'Dune', 'copies': -1}
rejected: {'title': 'Dune', 'copies': 1, 'admin': True}

1. 加长度百与百一书名。
2. 测试数量零与百。
3. 解释验证与授权区别。

先运行默认有限集成检查，跟踪 HTTP 至事务再返回，测试凭据随机生成且不打印。

下载 m14_capstone.py

"""Local catalogue capstone: SQLite, HTTP, scoped bearer tokens and idempotent loans.

Default: finite integration checks in a temporary directory.
To keep a local server running, set CS_CATALOGUE_TOKEN (32+ characters), then:
python m14_capstone.py --serve --db catalogue.sqlite --port 8000
This single-threaded http.server application is for local study, not public hosting.
"""
import argparse, json, os, re, secrets, sqlite3, tempfile
from contextlib import closing
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from threading import Thread
from urllib.error import HTTPError
from urllib.parse import parse_qs, urlsplit
from urllib.request import Request, urlopen

class Conflict(Exception):
    pass

class Catalogue:
    def __init__(self, path):
        self.path = path
        with closing(self.connect()) as db, db:
            db.executescript("""
              CREATE TABLE IF NOT EXISTS books(id INTEGER PRIMARY KEY, title TEXT NOT NULL,
                copies INTEGER NOT NULL CHECK(copies>=0), total INTEGER NOT NULL CHECK(copies<=total));
              CREATE TABLE IF NOT EXISTS loans(request_id TEXT PRIMARY KEY,
                book_id INTEGER NOT NULL REFERENCES books(id), member TEXT NOT NULL);
              CREATE INDEX IF NOT EXISTS title_index ON books(title);
            """)
            db.execute("INSERT OR IGNORE INTO books VALUES(1,'Dune',2,2)")
            db.execute("INSERT OR IGNORE INTO books VALUES(2,'Foundation',1,1)")
    def connect(self):
        db = sqlite3.connect(self.path, timeout=3)
        db.execute("PRAGMA foreign_keys=ON")
        return db
    def search(self, query):
        with closing(self.connect()) as db:
            # SQLite lower() has ASCII semantics here; literal substring, not LIKE wildcards.
            rows = db.execute("SELECT id,title,copies FROM books WHERE instr(lower(title),lower(?))>0 ORDER BY title,id LIMIT 100", (query,)).fetchall()
            return [dict(zip(("id", "title", "copies"), row)) for row in rows]
    def borrow(self, request_id, book_id, member):
        with closing(self.connect()) as db, db:
            db.execute("BEGIN IMMEDIATE")
            previous = db.execute("SELECT book_id,member FROM loans WHERE request_id=?", (request_id,)).fetchone()
            if previous:
                if previous != (book_id, member): raise Conflict("request ID reused with different inputs")
                return 200, {"request_id": request_id, "book_id": book_id, "member": member}
            if db.execute("UPDATE books SET copies=copies-1 WHERE id=? AND copies>0", (book_id,)).rowcount != 1:
                raise Conflict("book unavailable")
            db.execute("INSERT INTO loans VALUES(?,?,?)", (request_id, book_id, member))
            return 201, {"request_id": request_id, "book_id": book_id, "member": member}

class Handler(BaseHTTPRequestHandler):
    def setup(self):
        super().setup(); self.connection.settimeout(5)
    def log_message(self, *args):
        pass  # do not log credentials or request bodies
    def reply(self, status, value):
        body = json.dumps(value).encode("utf-8")
        self.send_response(status)
        if status == 401: self.send_header("WWW-Authenticate", 'Bearer realm="catalogue"')
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers(); self.wfile.write(body)
    def identity(self, scope):
        header = self.headers.get("Authorization", "")
        if not header.startswith("Bearer "):
            self.reply(401, {"error": "authentication required"}); return None
        supplied = header[7:].encode("utf-8")
        for token, (member, scopes) in self.server.tokens.items():
            if secrets.compare_digest(token.encode("utf-8"), supplied):
                if scope not in scopes:
                    self.reply(403, {"error": "permission denied"}); return None
                return member
        self.reply(401, {"error": "invalid credential"}); return None
    def do_GET(self):
        if self.identity("read") is None: return
        url = urlsplit(self.path)
        if url.path != "/books": self.reply(404, {"error": "unknown route"}); return
        params = parse_qs(url.query, keep_blank_values=True)
        if set(params) - {"q"} or len(params.get("q", [])) > 1:
            self.reply(400, {"error": "only one q parameter allowed"}); return
        query = params.get("q", [""])[0].strip()
        if len(query) > 100 or ("q" in params and not query):
            self.reply(400, {"error": "invalid query"}); return
        self.reply(200, self.server.catalogue.search(query))
    def do_POST(self):
        member = self.identity("borrow")
        if member is None: return
        if self.path != "/loans": self.reply(404, {"error": "unknown route"}); return
        if self.headers.get("Content-Type", "").split(";")[0].strip() != "application/json":
            self.reply(415, {"error": "JSON required"}); return
        if self.headers.get("Transfer-Encoding"):
            self.reply(400, {"error": "chunked bodies not supported"}); return
        lengths = self.headers.get_all("Content-Length", [])
        try:
            if len(lengths) != 1: raise ValueError("one body length required")
            length = int(lengths[0])
            if not 1 <= length <= 1024: raise ValueError("body length out of range")
            raw = self.rfile.read(length)
            if len(raw) != length: raise ValueError("truncated body")
            record = json.loads(raw.decode("utf-8"))
            if not isinstance(record, dict) or set(record) != {"request_id", "book_id"}: raise ValueError("invalid fields")
            if type(record["book_id"]) is not int or record["book_id"] <= 0: raise ValueError("positive integer ID required")
            if not isinstance(record["request_id"], str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", record["request_id"]): raise ValueError("invalid request ID")
        except (ValueError, UnicodeError, TimeoutError):
            self.reply(400, {"error": "invalid body"}); return
        try:
            status, value = self.server.catalogue.borrow(record["request_id"], record["book_id"], member)
            self.reply(status, value)
        except Conflict as error:
            self.reply(409, {"error": str(error)})
        except sqlite3.OperationalError:
            self.reply(503, {"error": "database unavailable"})

def make_server(path, tokens, port=0):
    catalogue = Catalogue(path)
    server = HTTPServer(("127.0.0.1", port), Handler)
    server.catalogue, server.tokens = catalogue, tokens
    return server

def self_test():
    token, guest = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
    with tempfile.TemporaryDirectory() as folder:
        path = str(Path(folder) / "catalogue.sqlite")
        server = make_server(path, {token: ("reader", {"read", "borrow"}), guest: ("guest", {"read"})})
        thread = Thread(target=server.serve_forever); thread.start()
        base = f"http://127.0.0.1:{server.server_port}"
        def request(route, body=None, credential=token):
            headers = {"Content-Type": "application/json"}
            if credential: headers["Authorization"] = "Bearer " + credential
            req = Request(base + route, data=json.dumps(body).encode() if body is not None else None, headers=headers)
            try:
                with urlopen(req, timeout=5) as response: return response.status, json.load(response)
            except HTTPError as error:
                with error: return error.code, json.load(error)
        try:
            assert request("/books", credential=None)[0] == 401
            assert request("/loans", {"request_id":"a", "book_id":1}, guest)[0] == 403
            assert request("/books?q=Dune")[1][0]["copies"] == 2
            assert request("/books?q=%27%20OR%201%3D1%20--")[1] == []
            assert request("/loans", {"request_id":"bad", "book_id":True})[0] == 400
            first = request("/loans", {"request_id":"a", "book_id":1})
            replay = request("/loans", {"request_id":"a", "book_id":1})
            assert first[0] == 201 and replay[0] == 200 and first[1] == replay[1]
            assert request("/loans", {"request_id":"a", "book_id":2})[0] == 409
            assert request("/books?q=Dune")[1][0]["copies"] == 1
            assert request("/loans", {"request_id":"b", "book_id":1})[0] == 201
            assert request("/loans", {"request_id":"c", "book_id":1})[0] == 409
            print("PASS: authentication, scopes, validation, bound SQL, idempotent replay and stock limits")
        finally:
            server.shutdown(); server.server_close(); thread.join(timeout=5); assert not thread.is_alive()
        reopened = Catalogue(path)
        assert reopened.search("Dune")[0]["copies"] == 0
        with closing(reopened.connect()) as db:
            total, copies = db.execute("SELECT total,copies FROM books WHERE id=1").fetchone()
            loans = db.execute("SELECT COUNT(*) FROM loans WHERE book_id=1").fetchone()[0]
            assert total == copies + loans == 2
        print("PASS: restart persistence and inventory invariant")

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--serve", action="store_true")
    parser.add_argument("--db", default="catalogue.sqlite")
    parser.add_argument("--port", type=int, default=8000)
    args = parser.parse_args()
    if not args.serve: self_test(); return
    token = os.environ.get("CS_CATALOGUE_TOKEN", "")
    if len(token) < 32: parser.error("set CS_CATALOGUE_TOKEN to a random 32+ character credential")
    server = make_server(args.db, {token: ("reader", {"read", "borrow"})}, args.port)
    print(f"Local catalogue: http://127.0.0.1:{server.server_port}/books (Bearer credential required)")
    try: server.serve_forever()
    except KeyboardInterrupt: pass
    finally: server.server_close()

if __name__ == "__main__": main()

PASS: authentication, scopes, validation, bound SQL, idempotent replay and stock limits
PASS: restart persistence and inventory invariant

1. 检查首次与重放借阅状态及相同体。
2. 扩展空查询、重复查询及跨身份 ID 碰撞测试。
3. 本地持久运行：把 CS_CATALOGUE_TOKEN 设为至少 32 字符随机凭据，再用 --serve --db catalogue.sqlite --port 8000，GET /books 附 Bearer 认证头。
4. 设计具有独立幂等键与事务的归还操作。

先尝试，再展开解答。★ 应用概念；★★ 结合概念；★★★ 进行设计或证明。

选择答案查看反馈，重置后可重做。无需 JavaScript 也可阅读答案表。

• OWASP 授权指南 — 阅读默认拒绝与逐请求权限检查，指出项目边界。
• OWASP SQL 注入防护 — 阅读参数查询例子并区分验证。

你已从表示与算法到操作系统、网络与持久化构建研究系统。展示项目约定、模块图、不变式、失败案例及证据限制，再在保留保证下扩展归还或多作者。