Langlebige API-Tokens fuer externe Zugriffe (Home Assistant)

Die Einkaufsliste war ueber /shopping-list und /shopping-list/groups schon
abrufbar, aber nur mit einem Login-JWT, das nach 7 Tagen ablaeuft - fuer HA
unpraktisch.

- Neue Tabelle api_tokens (Name, SHA-256-Hash, Ersteller, zuletzt genutzt).
  Der Klartext wird nur einmal beim Anlegen zurueckgegeben.
- get_current_user akzeptiert jetzt JWT ODER API-Token im Authorization-Header;
  bei Nutzung wird last_used_at fortgeschrieben.
- Endpunkte /api-tokens (Admin): auflisten, erzeugen, widerrufen.
- Einstellungsseite: Tokens verwalten inkl. einmaliger Anzeige und einem
  aufklappbaren Konfigurationsbeispiel fuer Home Assistant.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-22 16:36:17 +02:00
parent 93e487a1b0
commit faa9d6a25d
8 changed files with 240 additions and 7 deletions

View File

@@ -2,9 +2,11 @@ from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from datetime import datetime, timezone
from .database import get_db from .database import get_db
from .models import Role, User from .models import ApiToken, Role, User
from .security import decode_access_token from .security import decode_access_token, hash_api_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
@@ -18,9 +20,9 @@ _credentials_exc = HTTPException(
def get_current_user( def get_current_user(
token: str = Depends(oauth2_scheme), db: Session = Depends(get_db) token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)
) -> User: ) -> User:
"""Akzeptiert ein JWT aus dem Login oder ein langlebiges API-Token."""
payload = decode_access_token(token) payload = decode_access_token(token)
if payload is None: if payload is not None:
raise _credentials_exc
username = payload.get("sub") username = payload.get("sub")
if not username: if not username:
raise _credentials_exc raise _credentials_exc
@@ -29,6 +31,19 @@ def get_current_user(
raise _credentials_exc raise _credentials_exc
return user return user
# Kein gültiges JWT als API-Token prüfen (z.B. Home Assistant).
entry = (
db.query(ApiToken).filter(ApiToken.token_hash == hash_api_token(token)).first()
)
if entry is None:
raise _credentials_exc
user = db.get(User, entry.user_id) if entry.user_id else None
if user is None:
raise _credentials_exc
entry.last_used_at = datetime.now(timezone.utc)
db.commit()
return user
def require_admin(current_user: User = Depends(get_current_user)) -> User: def require_admin(current_user: User = Depends(get_current_user)) -> User:
if current_user.role != Role.admin: if current_user.role != Role.admin:

View File

@@ -7,6 +7,7 @@ from sqlalchemy import text
from .config import get_settings from .config import get_settings
from .database import Base, SessionLocal, engine from .database import Base, SessionLocal, engine
from .routers import ( from .routers import (
api_tokens,
auth, auth,
groups, groups,
locations, locations,
@@ -91,4 +92,5 @@ app.include_router(groups.router)
app.include_router(units.router) app.include_router(units.router)
app.include_router(views.router) app.include_router(views.router)
app.include_router(transfer.router) app.include_router(transfer.router)
app.include_router(api_tokens.router)
app.include_router(settings_router.router) app.include_router(settings_router.router)

View File

@@ -146,6 +146,25 @@ class Product(Base):
) )
class ApiToken(Base):
"""Langlebiges Token für externe Zugriffe (z.B. Home Assistant).
Gespeichert wird nur der SHA-256-Hash; der Klartext wird einmalig beim
Anlegen angezeigt.
"""
__tablename__ = "api_tokens"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(120), nullable=False)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
user_id: Mapped[int | None] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=True
)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class Barcode(Base): class Barcode(Base):
"""Zusätzliche EAN-Codes für ein Produkt ODER eine Gruppe. """Zusätzliche EAN-Codes für ein Produkt ODER eine Gruppe.

View File

@@ -0,0 +1,55 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from ..database import get_db
from ..deps import require_admin
from ..models import ApiToken, User
from ..schemas import ApiTokenCreate, ApiTokenCreated, ApiTokenOut
from ..security import generate_api_token, hash_api_token
router = APIRouter(prefix="/api-tokens", tags=["api-tokens"])
@router.get("", response_model=list[ApiTokenOut])
def list_tokens(
db: Session = Depends(get_db), _: User = Depends(require_admin)
) -> list[ApiToken]:
return db.query(ApiToken).order_by(ApiToken.id).all()
@router.post("", response_model=ApiTokenCreated, status_code=status.HTTP_201_CREATED)
def create_token(
payload: ApiTokenCreate,
db: Session = Depends(get_db),
admin: User = Depends(require_admin),
) -> ApiTokenCreated:
"""Erzeugt ein langlebiges Token. Der Klartext wird nur hier zurückgegeben."""
plain = generate_api_token()
entry = ApiToken(
name=payload.name.strip(),
token_hash=hash_api_token(plain),
user_id=admin.id,
)
db.add(entry)
db.commit()
db.refresh(entry)
return ApiTokenCreated(
id=entry.id,
name=entry.name,
created_at=entry.created_at,
last_used_at=entry.last_used_at,
token=plain,
)
@router.delete("/{token_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_token(
token_id: int,
db: Session = Depends(get_db),
_: User = Depends(require_admin),
) -> None:
entry = db.get(ApiToken, token_id)
if entry is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Token nicht gefunden")
db.delete(entry)
db.commit()

View File

@@ -23,6 +23,23 @@ class UnitCreate(BaseModel):
factor: float = Field(gt=0) factor: float = Field(gt=0)
# ---- API-Tokens (externe Zugriffe, z.B. Home Assistant) ----
class ApiTokenOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
created_at: datetime
last_used_at: datetime | None = None
class ApiTokenCreate(BaseModel):
name: str = Field(min_length=1, max_length=120)
class ApiTokenCreated(ApiTokenOut):
token: str # nur einmalig beim Anlegen
# ---- Barcodes ---- # ---- Barcodes ----
class BarcodeOut(BaseModel): class BarcodeOut(BaseModel):
model_config = ConfigDict(from_attributes=True) model_config = ConfigDict(from_attributes=True)

View File

@@ -1,3 +1,5 @@
import hashlib
import secrets
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
import bcrypt import bcrypt
@@ -30,6 +32,16 @@ def create_access_token(subject: str, role: str) -> str:
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm) return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def generate_api_token() -> str:
"""Erzeugt ein langlebiges Token für externe Zugriffe."""
return secrets.token_urlsafe(32)
def hash_api_token(token: str) -> str:
"""SHA-256 schnell genug, um bei jeder Anfrage geprüft zu werden."""
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def decode_access_token(token: str) -> dict | None: def decode_access_token(token: str) -> dict | None:
try: try:
return jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) return jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])

View File

@@ -158,6 +158,11 @@ export const api = {
updateUser: (id, body) => request(`/users/${id}`, { method: "PATCH", body }), updateUser: (id, body) => request(`/users/${id}`, { method: "PATCH", body }),
deleteUser: (id) => request(`/users/${id}`, { method: "DELETE" }), deleteUser: (id) => request(`/users/${id}`, { method: "DELETE" }),
// API-Tokens (externe Zugriffe)
listApiTokens: () => request("/api-tokens"),
createApiToken: (body) => request("/api-tokens", { method: "POST", body }),
deleteApiToken: (id) => request(`/api-tokens/${id}`, { method: "DELETE" }),
// Einstellungen // Einstellungen
listSettings: () => request("/settings"), listSettings: () => request("/settings"),
setSetting: (key, value) => setSetting: (key, value) =>

View File

@@ -82,6 +82,114 @@ export default function Settings() {
<button className="btn primary" disabled={busy}>{busy ? "Speichern" : "Speichern"}</button> <button className="btn primary" disabled={busy}>{busy ? "Speichern" : "Speichern"}</button>
</form> </form>
<ApiTokensCard />
</div> </div>
); );
} }
/** Langlebige Tokens für externe Zugriffe (Home Assistant, Skripte). */
function ApiTokensCard() {
const [tokens, setTokens] = useState([]);
const [name, setName] = useState("");
const [fresh, setFresh] = useState(null);
const [error, setError] = useState(null);
async function load() {
try {
setTokens(await api.listApiTokens());
} catch (err) {
setError(err.message);
}
}
useEffect(() => { load(); }, []);
async function create(e) {
e.preventDefault();
setError(null);
try {
const created = await api.createApiToken({ name });
setFresh(created.token);
setName("");
load();
} catch (err) {
setError(err.message);
}
}
async function remove(token) {
if (!confirm(`Token "${token.name}" widerrufen? Zugriffe damit funktionieren sofort nicht mehr.`)) return;
try {
await api.deleteApiToken(token.id);
load();
} catch (err) {
setError(err.message);
}
}
return (
<section className="card">
<div className="card-head"><Icon name="users" /><h2>API-Zugriff (z.B. Home Assistant)</h2></div>
{error && <div className="alert error"><Icon name="alert" size={16} />{error}</div>}
<p className="muted small mt-0">
Mit einem Token lässt sich die Einkaufsliste dauerhaft abfragen, ohne dass eine
Anmeldung abläuft. Der Token wird nur einmal angezeigt danach ist nur noch
sein Fingerabdruck gespeichert.
</p>
{fresh && (
<div className="alert ok">
<Icon name="check" size={16} />
<span>
Neuer Token bitte jetzt kopieren:
<div><code className="strong" style={{ wordBreak: "break-all" }}>{fresh}</code></div>
</span>
</div>
)}
<ul className="simple-list">
{tokens.map((t) => (
<li key={t.id}>
<span>
<strong>{t.name}</strong>
<span className="muted small">
{" "}· angelegt {new Date(t.created_at).toLocaleDateString("de-DE")}
{t.last_used_at
? ` · zuletzt genutzt ${new Date(t.last_used_at).toLocaleDateString("de-DE")}`
: " · noch nicht genutzt"}
</span>
</span>
<button className="btn-icon danger" onClick={() => remove(t)} title="Widerrufen">
<Icon name="trash" size={16} />
</button>
</li>
))}
{tokens.length === 0 && <li className="muted small">Noch keine Tokens.</li>}
</ul>
<form className="field-inline" onSubmit={create}>
<input placeholder="Name, z.B. Home Assistant" value={name}
onChange={(e) => setName(e.target.value)} required />
<button className="btn"><Icon name="plus" size={16} />Token erzeugen</button>
</form>
<details style={{ marginTop: "var(--sp-3)" }}>
<summary className="muted small">Beispiel für Home Assistant</summary>
<pre style={{ overflowX: "auto", fontSize: "0.78rem" }}>{`# configuration.yaml
rest:
- resource: http://DEIN-SERVER:8080/api/shopping-list
scan_interval: 900
headers:
Authorization: "Bearer DEIN_TOKEN"
sensor:
- name: "Einkaufsliste Anzahl"
value_template: "{{ value_json | length }}"
json_attributes_path: "$"
- name: "Einkaufsliste"
value_template: "{{ value_json | map(attribute='name') | join(', ') }}"`}</pre>
</details>
</section>
);
}