WebSocket Protecting
The WebSocket protocol doesn’t handle authorization or authentication. Practically, this means that a WebSocket opened from a page behind auth doesn’t "automatically" receive any sort of auth. You need to take steps to also secure the WebSocket connection.
Since you cannot customize WebSocket headers from JavaScript, you’re limited to the "implicit" auth (i.e. Basic or cookies) that’s sent from the browser. The more common approach to generates a token from your normal HTTP server and then have the client send the token (either as a query string in the WebSocket path or as the first WebSocket message). The WebSocket server then validates that the token is valid.
Note: Change all IP address to your localhost
Here is an example of how you authorize from query URL:
from fastapi import FastAPI, WebSocket, Depends, Request, HTTPException, Query
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
from pydantic import BaseModel
app = FastAPI()
class User(BaseModel):
username: str
password: str
class Settings(BaseModel):
authjwt_secret_key: str = "secret"
@AuthJWT.load_config
def get_config():
return Settings()
@app.exception_handler(AuthJWTException)
def authjwt_exception_handler(request: Request, exc: AuthJWTException):
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.message}
)
html = """
<!DOCTYPE html>
<html>
<head>
<title>Authorize</title>
</head>
<body>
<h1>WebSocket Authorize</h1>
<p>Token:</p>
<textarea id="token" rows="4" cols="50"></textarea><br><br>
<button onclick="websocketfun()">Send</button>
<ul id='messages'>
</ul>
<script>
const websocketfun = () => {
let token = document.getElementById("token").value
let ws = new WebSocket(`ws://192.168.18.202:8000/ws?token=${token}`)
ws.onmessage = (event) => {
let messages = document.getElementById('messages')
let message = document.createElement('li')
let content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
}
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket('/ws')
async def websocket(websocket: WebSocket, token: str = Query(...), Authorize: AuthJWT = Depends()):
await websocket.accept()
try:
Authorize.jwt_required("websocket",token=token)
# Authorize.jwt_optional("websocket",token=token)
# Authorize.jwt_refresh_token_required("websocket",token=token)
# Authorize.fresh_jwt_required("websocket",token=token)
await websocket.send_text("Successfully Login!")
decoded_token = Authorize.get_raw_jwt(token)
await websocket.send_text(f"Here your decoded token: {decoded_token}")
except AuthJWTException as err:
await websocket.send_text(err.message)
await websocket.close()
@app.post('/login')
def login(user: User, Authorize: AuthJWT = Depends()):
if user.username != "test" or user.password != "test":
raise HTTPException(status_code=401,detail="Bad username or password")
access_token = Authorize.create_access_token(subject=user.username,fresh=True)
refresh_token = Authorize.create_refresh_token(subject=user.username)
return {"access_token": access_token, "refresh_token": refresh_token}
You can copy the token from endpoint /login and then send them:
And your WebSocket route will respond back if the token is valid or not:
Here is an example of how you authorize from cookie:
from fastapi import FastAPI, WebSocket, Depends, Query
from fastapi.responses import HTMLResponse
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
from pydantic import BaseModel
app = FastAPI()
class Settings(BaseModel):
authjwt_secret_key: str = "secret"
authjwt_token_location: set = {"cookies"}
@AuthJWT.load_config
def get_config():
return Settings()
html = """
<!DOCTYPE html>
<html>
<head>
<title>Authorize</title>
</head>
<body>
<h1>WebSocket Authorize</h1>
<button onclick="websocketfun()">Send</button>
<ul id='messages'>
</ul>
<script>
const getCookie = (name) => {
const value = `; ${document.cookie}`;
const parts = value.split(`; ${name}=`);
if (parts.length === 2) return parts.pop().split(';').shift();
}
const websocketfun = () => {
let csrf_token = getCookie("csrf_access_token")
let ws = new WebSocket(`ws://192.168.18.202:8000/ws?csrf_token=${csrf_token}`)
ws.onmessage = (event) => {
let messages = document.getElementById('messages')
let message = document.createElement('li')
let content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
}
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket('/ws')
async def websocket(websocket: WebSocket, csrf_token: str = Query(...), Authorize: AuthJWT = Depends()):
await websocket.accept()
try:
Authorize.jwt_required("websocket",websocket=websocket,csrf_token=csrf_token)
# Authorize.jwt_optional("websocket",websocket=websocket,csrf_token=csrf_token)
# Authorize.jwt_refresh_token_required("websocket",websocket=websocket,csrf_token=csrf_token)
# Authorize.fresh_jwt_required("websocket",websocket=websocket,csrf_token=csrf_token)
await websocket.send_text("Successfully Login!")
decoded_token = Authorize.get_raw_jwt()
await websocket.send_text(f"Here your decoded token: {decoded_token}")
except AuthJWTException as err:
await websocket.send_text(err.message)
await websocket.close()
@app.get('/get-cookie')
def get_cookie(Authorize: AuthJWT = Depends()):
access_token = Authorize.create_access_token(subject='test',fresh=True)
refresh_token = Authorize.create_refresh_token(subject='test')
Authorize.set_access_cookies(access_token)
Authorize.set_refresh_cookies(refresh_token)
return {"msg":"Successfully login"}
You will see a simple page like this:
You can get the token from URL /get-cookie:
And click button send then your WebSocket route will respond back if the cookie and csrf token is match or cookie is valid or not: