FastAPI RabbitMQ Http auth backend

enabled_plugins

[rabbitmq_management,rabbitmq_auth_backend_cache,rabbitmq_auth_backend_http,rabbitmq_prometheus].

rabbitmq.conf

loopback_users.guest = false
listeners.tcp.default = 5672
management.listener.port = 15672
management.listener.ssl = false

#auth_backends.1 = http

## This configures rabbitmq_auth_backend_cache that delegates to
## the HTTP backend. If using this, make sure to comment the
## auth_backends.1 line above.
##
 auth_backends.1 = cache
#
 auth_cache.cached_backend = http
 auth_cache.cache_ttl = 10000


auth_http.http_method   = get
auth_http.user_path     = http://<yourhost - 'localhost' or for docker '172.17.0.1'>:8000/auth/user
auth_http.vhost_path    = http://<yourhost - 'localhost' or for docker '172.17.0.1'>:8000/auth/vhost
auth_http.resource_path = http://<yourhost - 'localhost' or for docker '172.17.0.1'>:8000/auth/resource
auth_http.topic_path    = http://<yourhost - 'localhost' or for docker '172.17.0.1'>:8000/auth/topic

rabbitmq_auth_backend_http_fastapi.py

# This is a very simple, incomplete template to write your own HTTP auth backend for RabbitMQ using FastAPI.
# TODO: Check username & passwords using FastAPI built in functions.
# TODO: Add Redis backend to store and retrive user info.
# !! Caution: This is a test app, Plesae add your own auth mechanism before using it. !!
# Author: github.com/Thermoflux

import datetime
import json
import os
from fastapi import FastAPI
from fastapi import Depends, FastAPI, Response, status, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import uvicorn


try:
    import thread
except ImportError:
    import _thread as thread


HOST = '0.0.0.0'
app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/auth/user")
async def authUser(request: Request, response: Response):
    print('*********authUser************')
    username = request.query_params.get('username')
    password = request.query_params.get('password')
    
    print(username + ' ' + password)
    response.status_code = 200
    response.body = b'allow'
    if username == 'admin':
        response.body = b'allow administrator'
    return response
    
@app.get("/auth/vhost")
# @limiter.limit("50/minute")
async def authLevelVhost(request: Request, response: Response):
    print('*********authLevelVhost************')
    username = request.query_params.get('username')
    ip = request.query_params.get('ip')
    tags = request.query_params.get('tags')
    print(username + ' ' + ip + '  ' + tags)
    response.status_code = 200
    response.body = b'allow'
    return response


@app.get("/auth/resource")
# @limiter.limit("50/minute")
async def authLevelResource(request: Request, response: Response):
    print('*********authLevelResource************')
    username = request.query_params.get('username')
    vhost = request.query_params.get('vhost')
    resource = request.query_params.get('resource')
    name = request.query_params.get('name')
    permission = request.query_params.get('permission')
    tags = request.query_params.get('tags')
    print(username + ' ' + vhost + '  ' + resource + '  ' + name + '  ' + permission + '  ' + tags)
    response.status_code = 200
    response.body = b'allow'
    return response
    
@app.get("/auth/topic")
# @limiter.limit("50/minute")
async def authLevelTopic(request: Request, response: Response):
    print('*********authLevelTopic************')
    username = request.query_params.get('username')
    vhost = request.query_params.get('vhost')
    resource = request.query_params.get('resource')
    name = request.query_params.get('name')
    permission = request.query_params.get('permission')
    tags = request.query_params.get('tags')
    print(username + ' ' + vhost + '  ' + resource + '  ' + name + '  ' + permission + '  ' + tags)
    response.status_code = 200
    response.body = b'allow'
    return response
    

if __name__ == '__main__':
        print("Running without SSL Certs")
        print("*****!! Make sure you edit the code to add your own auth mechanism !!*****")
        uvicorn.run(app, host=HOST, port=8000,debug=True)