1. Introduction
FastAPI is a modern, fast web framework for building API using Python. It is built on top of Starlette for the web parts and Pydantic for the data parts. FastAPI provides a built-in support for validation, serialization, and type hints. In this article, we will discuss about how to implement request anti-fraud and security in FastAPI.
2. Prevention of Request Forgery Attacks
Request Forgery attacks, also known as CSRF attacks, occur when an attacker sends a request on behalf of a user without their consent or knowledge. FastAPI provides an extension called CSRFProtect to guard against such attacks.
2.1. Installing CSRFProtect
The CSRFProtect extension can be installed using the following command:
pip install aiohttp-csrf
2.2. Usage of CSRFProtect
To use the CSRFProtect extension, we need to:
Instantiate the CSRFProtect object in the FastAPI app
Register the FastAPI middleware
Add the csrf_token to the HTML form
Here's an example:
from fastapi import FastAPI, Form
from aiohttp_csrf import csrf_protect_middleware, csrf_token_from_request
app = FastAPI()
app.add_middleware(csrf_protect_middleware)
@app.post('/submit-form/')
async def submit_form(csrf_token: str = Form(...), data: str = Form(...)):
csrf_token_from_request(request) # validate the csrf token
# process the form
return {'success': True}
In the above example, the CSRFProtect middleware is added as a middleware to the app using app.add_middleware(csrf_protect_middleware). This protects all the routes in the app from CSRF attacks. The csrf_token is added as a form field to the HTML form. And, in the FastAPI route, csrf_token_from_request(request) validates the token sent by the user.
3. Preventing SQL Injection Attacks
SQL Injection attacks occur when an attacker injects malicious SQL code into the SQL statements used by the application. This can cause the application to perform unintended actions. FastAPI provides support for preventing SQL Injection attacks using SQLAlchemy ORM and Prepared Statements.
3.1. Usage of SQLAlchemy ORM
The SQLAlchemy ORM provides an abstraction layer to the database. It allows us to write Pythonic code for interacting with the database. SQLAlchemy ORM also has built-in support for Prepared Statements which can help prevent SQL Injection attacks. Here's an example:
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
from database import SessionLocal, engine
from models import User
app = FastAPI()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post('/add-user/')
async def add_user(user: User, db: Session = Depends(get_db)):
stmt = text('INSERT INTO users (username, password) VALUES (:username, :password)')
stmt = stmt.bindparams(username=user.username, password=user.password)
db.execute(stmt)
db.commit()
return {'success': True}
In the above example, we use the SQLAlchemy ORM to insert a user into the database. We create the SQL statement using text() and bind the parameters using bindparams(). The db.execute() method safely executes the SQL statement and the db.commit() method commits the changes to the database.
3.2. Usage of Prepared Statements
Prepared Statements are a way to pre-compile a SQL statement and store it on the server. When the statement needs to be executed, only the parameters need to be sent. Prepared Statements prevent SQL Injection attacks by ensuring that the SQL statement and the parameters are separate. FastAPI provides support for Prepared Statements using the third-party aiohttp-sqlalchemy extension. You can install it using the following command:
pip install aiohttp-sqlalchemy
Here's an example:
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.sql import text
from sqlalchemy.ext.declarative import declarative_base
from aiohttp_sqlalchemy import SQLAlchemyMiddleware, register_session
from models import User
app = FastAPI()
# Database settings
DATABASE_URL = 'postgresql+asyncpg://user:password@localhost/dbname'
engine = create_async_engine(DATABASE_URL, echo=True, future=True, connect_args={'connect_timeout': 5})
# Middleware
Base = declarative_base()
app.middlewares.append(SQLAlchemyMiddleware(db_url=DATABASE_URL, engine=engine, base=Base))
# Dependency
async def get_session() -> AsyncSession:
async with register_session(engine) as session:
yield session
@app.post('/add-user/')
async def add_user(user: User, session: AsyncSession = Depends(get_session)):
stmt = text('INSERT INTO users (username, password) VALUES (:username, :password)')
stmt = stmt.bindparams(username=user.username, password=user.password)
await session.execute(stmt)
await session.commit()
return {'success': True}
In the above example, we use Prepared Statements to insert a user into the database. The aiohttp_sqlalchemy.SQLAlchemyMiddleware middleware is used to provide the SQLAlchemy session. We use the text() method to create the SQL statement and bind the parameters using bindparams(). The await session.execute() method safely executes the SQL statement and the await session.commit() method commits the changes to the database.
4. Conclusion
In this article, we learned how to implement request anti-fraud and security in FastAPI. We discussed how to prevent Request Forgery attacks using the CSRFProtect extension and how to prevent SQL Injection attacks using SQLAlchemy ORM and Prepared Statements. By following the best practices and using the built-in features of FastAPI, we can make our applications more secure and prevent malicious attacks.