ptarrant/api_test
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
Phillip Tarrant
2025-10-15 13:58:10 -05:00
commit 85d9214883
9 changed files with 300 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
/venv/
/data/
**/__pycache__/

16
app/app.py Normal file
View File

@@ -0,0 +1,16 @@
from flask import Flask
from blueprints.main import main_bp
from blueprints.api import api_bp
def create_app():
app = Flask(__name__)
# Register blueprints
app.register_blueprint(main_bp)
app.register_blueprint(api_bp, url_prefix="/api")
return app
if __name__ == "__main__":
app = create_app()
app.run(debug=True)

24
app/blueprints/api.py Normal file
View File

@@ -0,0 +1,24 @@
import os
from flask import Blueprint, jsonify
from flask_httpauth import HTTPTokenAuth
from utils.token_store import ApiKeyStore
api_bp = Blueprint("api", __name__)
auth = HTTPTokenAuth(scheme="Bearer")
STORE_PATH = os.getenv("API_TOKEN_FILE", "data/api_tokens.yaml")
store = ApiKeyStore(STORE_PATH, bcrypt_rounds=12)
@auth.verify_token
def verify_token(token: str):
return store.verify(token)
@auth.error_handler
def auth_error(status):
return jsonify({"error": "Unauthorized"}), status
@api_bp.route("/health", methods=["GET"])
@auth.login_required
def health():
return jsonify({"healthy": True})

7
app/blueprints/main.py Normal file
View File

@@ -0,0 +1,7 @@
from flask import Blueprint, render_template
main_bp = Blueprint("main", __name__)
@main_bp.route("/")
def index():
return render_template("index.html")

10
app/templates/index.html Normal file
View File

@@ -0,0 +1,10 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Hello World</title>
</head>
<body>
<h1>Hello, World!</h1>
</body>
</html>

218
app/utils/token_store.py Normal file
View File

@@ -0,0 +1,218 @@
# token_store.py
from __future__ import annotations
import os
import io
import uuid
import time
import json
import tempfile
import threading
import secrets
from dataclasses import dataclass, asdict
from typing import List, Optional, Dict
import bcrypt
import yaml
@dataclass
class ApiKey:
id: str
label: str
hash: str # bcrypt hash (utf-8 str)
created_at: str # ISO8601-ish seconds precision
active: bool = True
@staticmethod
def now_iso() -> str:
# keep it simple; no tz handling needed for a local file
return time.strftime("%Y-%m-%dT%H:%M:%S")
class ApiKeyStore:
"""
Minimal API key manager:
- YAML file on disk
- stores bcrypt hashes only
- mint/list/get/deactivate/delete/rotate/verify
File format:
---
version: 1
keys:
- id: <uuid4>
label: "build bot"
hash: "$2b$12$..."
created_at: "2025-10-15T10:22:00"
active: true
"""
def __init__(self, path: str, bcrypt_rounds: int = 12):
self.path = path
self.bcrypt_rounds = bcrypt_rounds
self._lock = threading.Lock()
self._data: Dict[str, object] = {"version": 1, "keys": []}
self._load_if_exists()
# ---------- public API ----------
def mint(self, label: str = "") -> Dict[str, str]:
"""
Create a new API token.
Returns a dict with plaintext `token` (show it once!) and its `id`.
"""
token = secrets.token_urlsafe(32)
hashed = bcrypt.hashpw(token.encode("utf-8"),
bcrypt.gensalt(rounds=self.bcrypt_rounds)).decode("utf-8")
key = ApiKey(
id=str(uuid.uuid4()),
label=label,
hash=hashed,
created_at=ApiKey.now_iso(),
active=True,
)
with self._lock:
keys = self._keys()
keys.append(key)
self._commit()
return {"id": key.id, "token": token}
def list(self, include_inactive: bool = True) -> List[Dict[str, object]]:
keys = self._keys()
out = []
for k in keys:
if include_inactive or k.active:
d = asdict(k)
d.pop("hash") # don’t leak hashes in listing
out.append(d)
return out
def get(self, key_id: str) -> Optional[Dict[str, object]]:
k = self._find(key_id)
if not k:
return None
d = asdict(k)
d.pop("hash")
return d
def deactivate(self, key_id: str) -> bool:
with self._lock:
k = self._find(key_id)
if not k:
return False
if not k.active:
return True
k.active = False
self._commit()
return True
def activate(self, key_id: str) -> bool:
with self._lock:
k = self._find(key_id)
if not k:
return False
if k.active:
return True
k.active = True
self._commit()
return True
def delete(self, key_id: str) -> bool:
with self._lock:
keys = self._keys()
before = len(keys)
keys[:] = [k for k in keys if k.id != key_id]
changed = len(keys) != before
if changed:
self._commit()
return changed
def rotate(self, key_id: str) -> Optional[Dict[str, str]]:
"""
Mint a brand-new token for an existing key ID (keeps label).
The old hash is replaced. Returns {"id", "token"} or None if not found.
"""
with self._lock:
k = self._find(key_id)
if not k:
return None
new_token = secrets.token_urlsafe(32)
new_hash = bcrypt.hashpw(new_token.encode("utf-8"),
bcrypt.gensalt(rounds=self.bcrypt_rounds)).decode("utf-8")
k.hash = new_hash
k.created_at = ApiKey.now_iso()
k.active = True
self._commit()
return {"id": k.id, "token": new_token}
def verify(self, token: str) -> bool:
"""
Check whether a plaintext token matches any ACTIVE key.
"""
if not token:
return False
token_b = token.encode("utf-8")
# no lock needed for read-only access; file commits are atomic
for k in self._keys():
if not k.active:
continue
try:
if bcrypt.checkpw(token_b, k.hash.encode("utf-8")):
return True
except ValueError:
# malformed hash in file—treat as non-match
continue
return False
# ---------- internals ----------
def _load_if_exists(self) -> None:
if not os.path.exists(self.path):
return
with open(self.path, "r", encoding="utf-8") as f:
raw = yaml.safe_load(f) or {}
version = raw.get("version", 1)
if version != 1:
raise ValueError(f"Unsupported token store version: {version}")
keys_raw = raw.get("keys", [])
keys: List[ApiKey] = []
for item in keys_raw:
keys.append(ApiKey(
id=str(item["id"]),
label=str(item.get("label", "")),
hash=str(item["hash"]),
created_at=str(item.get("created_at", ApiKey.now_iso())),
active=bool(item.get("active", True)),
))
self._data = {"version": 1, "keys": keys}
def _keys(self) -> List[ApiKey]:
return self._data["keys"] # type: ignore[return-value]
def _commit(self) -> None:
"""
Write YAML atomically to avoid partial writes.
"""
directory = os.path.dirname(os.path.abspath(self.path)) or "."
os.makedirs(directory, exist_ok=True)
# Convert dataclasses to serializable dicts
payload = {
"version": 1,
"keys": [asdict(k) for k in self._keys()],
# optional: a tiny meta line to help with diffs
"_meta": {"updated": ApiKey.now_iso(), "count": len(self._keys())},
}
tmp_fd, tmp_path = tempfile.mkstemp(prefix=".apikeystore.", dir=directory)
try:
with io.open(tmp_fd, "w", encoding="utf-8") as tmp:
yaml.safe_dump(payload, tmp, sort_keys=False)
os.replace(tmp_path, self.path) # atomic on POSIX & Windows
finally:
# If os.replace throws, ensure temp file is gone
if os.path.exists(tmp_path):
try:
os.remove(tmp_path)
except OSError:
pass

11
mint.py Normal file
View File

@@ -0,0 +1,11 @@
from app.utils.token_store import ApiKeyStore
store = ApiKeyStore("data/api_tokens.yaml")
# res = store.mint(label="local dev")
# print("NEW TOKEN (save this now):")
# print(res) # {"id": "...uuid...", "token": "...plaintext..."}
print("List keys (no hashes):")
print(store.list(include_inactive=True))
# print("Verify works?", store.verify(res["token"])) # True

10
requirements.txt Normal file
View File

@@ -0,0 +1,10 @@
bcrypt==5.0.0
blinker==1.9.0
click==8.3.0
Flask==3.1.2
Flask-HTTPAuth==4.8.0
itsdangerous==2.2.0
Jinja2==3.1.6
MarkupSafe==3.0.3
PyYAML==6.0.3
Werkzeug==3.1.3

1
run_flask.sh Executable file
View File

@@ -0,0 +1 @@
python app/app.py