Review

Verify KegWatch inventory webhook signatures

Craft-brewery POS now receives live keg-level webhooks from KegWatch sensors. Rejects unsigned payloads before enqueuing inventory apply jobs so spoofed empties cannot zero out taps mid-service.

PythonTier 3securitywebhookscrypto

Click a line to flag it, pick one or more labels, then submit. If the change looks correct, approve it.

services/webhooks/kegwatch.py+18-5

14	14		import hashlib
15	15		import hmac
16	16		from fastapi import Header, HTTPException, Request
17	17
18	18		from app.config import settings
19	19		from app.jobs import enqueue_keg_delta
20	20
	21	+	def _expected_signature(body: bytes) -> str:
	22	+	return hmac.new(
	23	+	settings.KEGWATCH_WEBHOOK_SECRET.encode("utf-8"),
	24	+	body,
	25	+	hashlib.sha256,
	26	+	).hexdigest()
	27	+
21		-	@router.post("/webhooks/kegwatch")
22		-	async def kegwatch_webhook(request: Request):
23		-	body = await request.json()
24		-	enqueue_keg_delta(body)
25		-	return {"ok": True}
	28	+	@router.post("/webhooks/kegwatch")
	29	+	async def kegwatch_webhook(
	30	+	request: Request,
	31	+	x_kegwatch_signature: str = Header(...),
	32	+	):
	33	+	raw = await request.body()
	34	+	expected = _expected_signature(raw)
	35	+	if x_kegwatch_signature != expected:
	36	+	raise HTTPException(status_code=401, detail="invalid signature")
	37	+	enqueue_keg_delta(await request.json())
	38	+	return {"ok": True}