bmondream/sms-remote
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add "typ" JWT claim, restrict device list to PRIMARY, rename endpoint routes

Browse Source
This commit is contained in:
bmondream
2026-05-01 19:51:20 +02:00
parent 1391e34e69
commit e5128df5a0
1 changed files with 15 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+15 -11
View File
@@ -1,7 +1,7 @@
from datetime import timedelta from datetime import timedelta
from flask import Flask, abort, jsonify, make_response, request from flask import Flask, abort, jsonify, make_response, request
from flask_bcrypt import Bcrypt from flask_bcrypt import Bcrypt
from flask_jwt_extended import JWTManager, create_access_token, create_refresh_token, get_jwt_identity, jwt_required from flask_jwt_extended import JWTManager, create_access_token, create_refresh_token, get_jwt, get_jwt_identity, jwt_required
import logging import logging
import sqlite3 import sqlite3
@@ -25,7 +25,7 @@ app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO) app.logger.setLevel(logging.INFO)
@app.route("/login", methods=["POST"]) @app.route("/api/v1/login", methods=["POST"])
def login(): def login():
access_key = request.json.get("access_key", None) access_key = request.json.get("access_key", None)
if access_key is None: if access_key is None:
@@ -35,35 +35,39 @@ def login():
if secret_key is None: if secret_key is None:
return make_response(jsonify({"msg": "Missing secret key"}), 400) return make_response(jsonify({"msg": "Missing secret key"}), 400)
res = cur.execute("SELECT access_key, secret_key_hash FROM devices WHERE access_key = ?", (access_key,)) res_tuple = cur.execute("SELECT secret_key_hash, type FROM devices WHERE access_key = ?", (access_key,)).fetchone()
res_tuple = res.fetchone()
if res_tuple is None: if res_tuple is None:
return make_response(jsonify({"msg": "Invalid access key or secret key"}), 401) return make_response(jsonify({"msg": "Invalid access key or secret key"}), 401)
secret_key_hash = res_tuple[1] (secret_key_hash, type_claim) = res_tuple
if flask_bcrypt.check_password_hash(secret_key_hash, secret_key): if flask_bcrypt.check_password_hash(secret_key_hash, secret_key):
# "typ" claim means device type
extra_claims = {"typ": type_claim}
# generate and return JWT # generate and return JWT
return make_response( return make_response(
jsonify({"access_token": create_access_token(identity=access_key), jsonify({"access_token": create_access_token(identity=access_key, additional_claims=extra_claims),
"refresh_token": create_refresh_token(identity=access_key)}), "refresh_token": create_refresh_token(identity=access_key, additional_claims=extra_claims)}),
200) 200)
else: else:
return make_response(jsonify({"msg": "Invalid access key or secret key"}), 401) return make_response(jsonify({"msg": "Invalid access key or secret key"}), 401)
@app.route("/verify-token", methods=["GET"]) @app.route("/api/v1/verify-token", methods=["GET"])
@jwt_required() @jwt_required()
def verify_token(): def verify_token():
return make_response(jsonify(logged_in_as=get_jwt_identity()), 200) return make_response(jsonify(logged_in_as=get_jwt_identity()), 200)
@app.route("/refresh-token", methods=["POST"]) @app.route("/api/v1/refresh-token", methods=["POST"])
@jwt_required(refresh=True) @jwt_required(refresh=True)
def refresh_token(): def refresh_token():
return make_response(jsonify({"access_token": create_access_token(identity=get_jwt_identity())}), 200) return make_response(jsonify({"access_token": create_access_token(identity=get_jwt_identity(), additional_claims={"typ": get_jwt()["typ"]})}), 200)
@app.route("/devices", methods=["GET"]) @app.route("/api/v1/devices", methods=["GET"])
@jwt_required() @jwt_required()
def get_all_devices(): def get_all_devices():
if get_jwt()["typ"] != "PRIMARY":
return make_response(jsonify(msg="Only a PRIMARY device can list all devices"), 403)
return make_response(jsonify([d.to_dict() for d in db.get_all_devices(cur)]), 200) return make_response(jsonify([d.to_dict() for d in db.get_all_devices(cur)]), 200)