Data Integrity Fundamentals

Introduction

Welcome to the very first lesson of the " Secure Data Handling and Integrity in FastAPI " course! In this lesson, we will explore the fundamentals of data integrity, a crucial aspect of secure data handling. Data integrity ensures that information remains accurate, consistent, and reliable throughout its lifecycle. By the end of this lesson, you'll understand the importance of data integrity and how to implement it in your FastAPI applications. Let's dive in! 🚀

Understanding Data Integrity

As we discussed earlier, if an attacker already has access to some endpoints or the network, it is still important to limit their capabilities. Previously, we focused on securing specific types of data that attackers can target, such as user credentials and JWT tokens. Now, let's consider a more generalized approach, focusing on the integrity of ANY DATA within the application. Data integrity is the assurance that data is accurate, consistent, and reliable. It plays a vital role in maintaining the trustworthiness of information, especially in applications where data is frequently accessed and modified. There are three main types of data integrity: Physical Integrity : Ensures that data is stored and retrieved without corruption due to hardware failures or environmental factors. Logical Integrity : Maintains the correctness and validity of data within a database, ensuring that it adheres to predefined rules and constraints. Referential Integrity : Ensures that relationships between data in different tables remain consistent, preventing orphaned records or invalid references. Understanding these types helps us appreciate the various ways data integrity can be compromised and the importance of safeguarding it. In this lesson, we primarily focus on logical and referential integrity. To better understand how logical and referential integrity apply in real-world FastAPI applications, consider a case where multiple related entities (e.g., users and their posts) exist in a database. Logical integrity ensures that a user's email follows a proper format and is unique, while referential integrity ensures that every post references a valid user ID. If a user is deleted, cascading deletes or foreign key constraints are often used to maintain referential integrity. These constraints are typically enforced at the database level (e.g., PostgreSQL, MySQL) and can be configured in SQLAlchemy models.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class SnippetRequest(BaseModel):
    content: str

@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    # Store content in the database without any integrity checks
    return {'message': 'Snippet stored', 'content': request.content}

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class SnippetRequest(BaseModel):
    content: str

@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    # Store content in the database without any integrity checks
    return {'message': 'Snippet stored', 'content': request.content}

Python

import hashlib

# Function to hash data
def hash_data(data: str) -> str:
    return hashlib.sha256(data.encode('utf-8')).hexdigest()

# Example of hashing snippet content
@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    hashed_content = hash_data(request.content)
    # Store hashed_content in the database
    return {'message': 'Snippet stored with hash', 'hash': hashed_content}

import hashlib

# Function to hash data
def hash_data(data: str) -> str:
    return hashlib.sha256(data.encode('utf-8')).hexdigest()

# Example of hashing snippet content
@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    hashed_content = hash_data(request.content)
    # Store hashed_content in the database
    return {'message': 'Snippet stored with hash', 'hash': hashed_content}

Python

def verify_data_integrity(content: str, stored_hash: str) -> bool:
    calculated_hash = hash_data(content)
    return calculated_hash == stored_hash

@app.get('/api/snippets/{snippet_id}')
async def get_snippet(snippet_id: str):
    # Retrieve snippet and its hash from the database
    # Example: snippet = db.get_snippet_by_id(snippet_id)
    snippet = {'content': 'example', 'hash': 'exampleHash'}  # Placeholder

    # Verify the integrity of the retrieved snippet
    is_valid = verify_data_integrity(snippet['content'], snippet['hash'])
    if is_valid:
        return snippet
    else:
        raise HTTPException(status_code=403, detail='Data integrity check failed')

def verify_data_integrity(content: str, stored_hash: str) -> bool:
    calculated_hash = hash_data(content)
    return calculated_hash == stored_hash

@app.get('/api/snippets/{snippet_id}')
async def get_snippet(snippet_id: str):
    # Retrieve snippet and its hash from the database
    # Example: snippet = db.get_snippet_by_id(snippet_id)
    snippet = {'content': 'example', 'hash': 'exampleHash'}  # Placeholder

    # Verify the integrity of the retrieved snippet
    is_valid = verify_data_integrity(snippet['content'], snippet['hash'])
    if is_valid:
        return snippet
    else:
        raise HTTPException(status_code=403, detail='Data integrity check failed')

Python

import hmac
import os

# Function to generate HMAC
def generate_hmac(data: str, secret: str) -> str:
    return hmac.new(
        secret.encode('utf-8'), 
        data.encode('utf-8'), 
        hashlib.sha256
    ).hexdigest()

# Example of generating HMAC for snippet content
@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    secret = os.getenv('HMAC_SECRET')
    hmac_value = generate_hmac(request.content, secret)
    # Store hmac_value alongside the snippet in the database
    return {'message': 'Snippet stored with HMAC', 'hmac': hmac_value}

import hmac
import os

# Function to generate HMAC
def generate_hmac(data: str, secret: str) -> str:
    return hmac.new(
        secret.encode('utf-8'), 
        data.encode('utf-8'), 
        hashlib.sha256
    ).hexdigest()

# Example of generating HMAC for snippet content
@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    secret = os.getenv('HMAC_SECRET')
    hmac_value = generate_hmac(request.content, secret)
    # Store hmac_value alongside the snippet in the database
    return {'message': 'Snippet stored with HMAC', 'hmac': hmac_value}

Python

# Function to verify HMAC
def verify_hmac(data: str, hmac_value: str, secret: str) -> bool:
    calculated_hmac = generate_hmac(data, secret)
    return hmac.compare_digest(calculated_hmac, hmac_value)

# Example of verifying snippet data
@app.get('/api/snippets/{snippet_id}')
async def get_snippet(snippet_id: str):
    secret = os.getenv('HMAC_SECRET')
    # Retrieve snippet and its HMAC from the database
    snippet = {'content': 'example', 'hmac': 'exampleHMAC'}  # Placeholder
    is_valid = verify_hmac(snippet['content'], snippet['hmac'], secret)
    if is_valid:
        return snippet
    else:
        raise HTTPException(status_code=403, detail='Invalid HMAC')

# Function to verify HMAC
def verify_hmac(data: str, hmac_value: str, secret: str) -> bool:
    calculated_hmac = generate_hmac(data, secret)
    return hmac.compare_digest(calculated_hmac, hmac_value)

# Example of verifying snippet data
@app.get('/api/snippets/{snippet_id}')
async def get_snippet(snippet_id: str):
    secret = os.getenv('HMAC_SECRET')
    # Retrieve snippet and its HMAC from the database
    snippet = {'content': 'example', 'hmac': 'exampleHMAC'}  # Placeholder
    is_valid = verify_hmac(snippet['content'], snippet['hmac'], secret)
    if is_valid:
        return snippet
    else:
        raise HTTPException(status_code=403, detail='Invalid HMAC')

Python

import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import serialization

# Function to generate digital signature
def generate_signature(data: str, private_key: str) -> str:
    private_key_obj = serialization.load_pem_private_key(
        private_key.encode('utf-8'), 
        password=None
    )
    signature = private_key_obj.sign(
        data.encode('utf-8'),
        padding.PKCS1v15(),
        hashes.SHA256()
    )
    return base64.b64encode(signature).decode('utf-8')

# Example: Generate signature for an admin action
private_key = 'examplePrivateKey'  # Placeholder - should be loaded securely
data = 'adminAction'
signature = generate_signature(data, private_key)
# The signature would be sent along with the request

import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import serialization

# Function to generate digital signature
def generate_signature(data: str, private_key: str) -> str:
    private_key_obj = serialization.load_pem_private_key(
        private_key.encode('utf-8'), 
        password=None
    )
    signature = private_key_obj.sign(
        data.encode('utf-8'),
        padding.PKCS1v15(),
        hashes.SHA256()
    )
    return base64.b64encode(signature).decode('utf-8')

# Example: Generate signature for an admin action
private_key = 'examplePrivateKey'  # Placeholder - should be loaded securely
data = 'adminAction'
signature = generate_signature(data, private_key)
# The signature would be sent along with the request

Python

# Function to verify digital signature
def verify_signature(data: str, signature: str, public_key: str) -> bool:
    try:
        public_key_obj = serialization.load_pem_public_key(public_key.encode('utf-8'))
        signature_bytes = base64.b64decode(signature.encode('utf-8'))
        public_key_obj.verify(
            signature_bytes,
            data.encode('utf-8'),
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return True
    except Exception:
        return False

# Example of verifying admin action
@app.get('/api/admin/test')
async def admin_test(request: Request):
    signature = request.headers.get('signature')
    public_key = 'examplePublicKey'  # Placeholder
    data = 'adminAction'
    is_valid = verify_signature(data, signature, public_key)
    if is_valid:
        return {'message': 'Admin action verified'}
    else:
        raise HTTPException(status_code=403, detail='Invalid signature')

# Function to verify digital signature
def verify_signature(data: str, signature: str, public_key: str) -> bool:
    try:
        public_key_obj = serialization.load_pem_public_key(public_key.encode('utf-8'))
        signature_bytes = base64.b64decode(signature.encode('utf-8'))
        public_key_obj.verify(
            signature_bytes,
            data.encode('utf-8'),
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return True
    except Exception:
        return False

# Example of verifying admin action
@app.get('/api/admin/test')
async def admin_test(request: Request):
    signature = request.headers.get('signature')
    public_key = 'examplePublicKey'  # Placeholder
    data = 'adminAction'
    is_valid = verify_signature(data, signature, public_key)
    if is_valid:
        return {'message': 'Admin action verified'}
    else:
        raise HTTPException(status_code=403, detail='Invalid signature')

Python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class SnippetRequest(BaseModel):
    content: str

@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    # Store content in the database without any integrity checks
    return {'message': 'Snippet stored', 'content': request.content}

Python

import hashlib

# Function to hash data
def hash_data(data: str) -> str:
    return hashlib.sha256(data.encode('utf-8')).hexdigest()

# Example of hashing snippet content
@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    hashed_content = hash_data(request.content)
    # Store hashed_content in the database
    return {'message': 'Snippet stored with hash', 'hash': hashed_content}

Python

def verify_data_integrity(content: str, stored_hash: str) -> bool:
    calculated_hash = hash_data(content)
    return calculated_hash == stored_hash

@app.get('/api/snippets/{snippet_id}')
async def get_snippet(snippet_id: str):
    # Retrieve snippet and its hash from the database
    # Example: snippet = db.get_snippet_by_id(snippet_id)
    snippet = {'content': 'example', 'hash': 'exampleHash'}  # Placeholder

    # Verify the integrity of the retrieved snippet
    is_valid = verify_data_integrity(snippet['content'], snippet['hash'])
    if is_valid:
        return snippet
    else:
        raise HTTPException(status_code=403, detail='Data integrity check failed')

Python

import hmac
import os

# Function to generate HMAC
def generate_hmac(data: str, secret: str) -> str:
    return hmac.new(
        secret.encode('utf-8'), 
        data.encode('utf-8'), 
        hashlib.sha256
    ).hexdigest()

# Example of generating HMAC for snippet content
@app.post('/api/snippets')
async def create_snippet(request: SnippetRequest):
    secret = os.getenv('HMAC_SECRET')
    hmac_value = generate_hmac(request.content, secret)
    # Store hmac_value alongside the snippet in the database
    return {'message': 'Snippet stored with HMAC', 'hmac': hmac_value}

Python

# Function to verify HMAC
def verify_hmac(data: str, hmac_value: str, secret: str) -> bool:
    calculated_hmac = generate_hmac(data, secret)
    return hmac.compare_digest(calculated_hmac, hmac_value)

# Example of verifying snippet data
@app.get('/api/snippets/{snippet_id}')
async def get_snippet(snippet_id: str):
    secret = os.getenv('HMAC_SECRET')
    # Retrieve snippet and its HMAC from the database
    snippet = {'content': 'example', 'hmac': 'exampleHMAC'}  # Placeholder
    is_valid = verify_hmac(snippet['content'], snippet['hmac'], secret)
    if is_valid:
        return snippet
    else:
        raise HTTPException(status_code=403, detail='Invalid HMAC')

Python

import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import serialization

# Function to generate digital signature
def generate_signature(data: str, private_key: str) -> str:
    private_key_obj = serialization.load_pem_private_key(
        private_key.encode('utf-8'), 
        password=None
    )
    signature = private_key_obj.sign(
        data.encode('utf-8'),
        padding.PKCS1v15(),
        hashes.SHA256()
    )
    return base64.b64encode(signature).decode('utf-8')

# Example: Generate signature for an admin action
private_key = 'examplePrivateKey'  # Placeholder - should be loaded securely
data = 'adminAction'
signature = generate_signature(data, private_key)
# The signature would be sent along with the request

Python

# Function to verify digital signature
def verify_signature(data: str, signature: str, public_key: str) -> bool:
    try:
        public_key_obj = serialization.load_pem_public_key(public_key.encode('utf-8'))
        signature_bytes = base64.b64decode(signature.encode('utf-8'))
        public_key_obj.verify(
            signature_bytes,
            data.encode('utf-8'),
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return True
    except Exception:
        return False

# Example of verifying admin action
@app.get('/api/admin/test')
async def admin_test(request: Request):
    signature = request.headers.get('signature')
    public_key = 'examplePublicKey'  # Placeholder
    data = 'adminAction'
    is_valid = verify_signature(data, signature, public_key)
    if is_valid:
        return {'message': 'Admin action verified'}
    else:
        raise HTTPException(status_code=403, detail='Invalid signature')

Introduction

Understanding Data Integrity

Join the 1M+ learners on CodeSignal

The Vulnerable Code

Implementing Data Hashing

Retrieving and Verifying the Snippet

Generating HMAC for Data Authentication

Implementing HMAC Verification

Generating Digital Signatures

Implementing Digital Signature Verification

Conclusion and Next Steps