0
# Additional Features
1

2
This document covers additional BlackSheep features including CORS, CSRF protection, sessions, OpenAPI documentation, security headers, compression, templating, and file handling utilities.
3

4
## CORS (Cross-Origin Resource Sharing)
5

6
BlackSheep provides comprehensive CORS support for handling cross-origin requests from web browsers.
7

8
### Basic CORS Setup
9

10
```python { .api }
11
from blacksheep import Application
12
from blacksheep.server.cors import CORSPolicy, CORSStrategy
13

14
app = Application()
15

16
# Simple CORS setup (allows all origins)
17
app.use_cors()
18

19
# Configure CORS with specific options
20
cors_strategy = app.use_cors(
21
    allow_origins=["https://example.com", "https://app.example.com"],
22
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
23
    allow_headers=["Content-Type", "Authorization", "X-Requested-With"],
24
    allow_credentials=True,
25
    max_age=3600,  # Cache preflight for 1 hour
26
    expose_headers=["X-Total-Count", "X-Page-Info"]
27
)
28
```
29

30
### Named CORS Policies
31

32
```python { .api }
33
# Multiple CORS policies for different endpoints
34
app.use_cors()  # Default policy
35

36
# API-specific policy
37
app.add_cors_policy("api", 
38
    allow_origins=["https://api-client.example.com"],
39
    allow_methods=["GET", "POST", "PUT", "DELETE"],
40
    allow_headers=["Authorization", "Content-Type"],
41
    allow_credentials=True
42
)
43

44
# Public API policy
45
app.add_cors_policy("public",
46
    allow_origins="*",
47
    allow_methods=["GET", "OPTIONS"],
48
    allow_headers=["Content-Type"],
49
    allow_credentials=False
50
)
51

52
# Widget embed policy
53
app.add_cors_policy("embed",
54
    allow_origins=["https://widget.example.com", "https://embed.example.com"],
55
    allow_methods=["GET", "POST"],
56
    allow_headers=["Content-Type", "X-Widget-Token"],
57
    expose_headers=["X-Widget-Version"]
58
)
59
```
60

61
### CORS Policy Configuration
62

63
```python { .api }
64
from blacksheep.server.cors import CORSPolicy
65

66
# Manual CORS policy creation
67
cors_policy = CORSPolicy(
68
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
69
    allow_headers=["Content-Type", "Authorization", "X-API-Key"],
70
    allow_origins=["https://trusted-domain.com"],
71
    allow_credentials=True,
72
    max_age=86400,  # 24 hours
73
    expose_headers=["X-Rate-Limit-Remaining", "X-Rate-Limit-Reset"]
74
)
75

76
# Advanced CORS strategy
77
strategy = CORSStrategy()
78
strategy.add_policy("default", cors_policy)
79

80
# Custom origin validation
81
def validate_origin(origin: str) -> bool:
82
    # Custom logic for validating origins
83
    allowed_patterns = ["*.example.com", "localhost:*"]
84
    return any(matches_pattern(origin, pattern) for pattern in allowed_patterns)
85

86
cors_policy = CORSPolicy(
87
    allow_origins=validate_origin,  # Function for dynamic validation
88
    allow_methods="*",
89
    allow_headers="*"
90
)
91
```
92

93
### Route-Specific CORS
94

95
```python { .api }
96
from blacksheep.server.cors import cors
97

98
# Apply CORS to specific routes
99
@app.get("/api/public")
100
@cors("public")  # Use "public" CORS policy
101
async def public_api():
102
    return json({"public": True})
103

104
@app.get("/api/private") 
105
@cors("api")  # Use "api" CORS policy
106
async def private_api():
107
    return json({"private": True})
108

109
# Route without CORS (uses default if set)
110
@app.get("/internal")
111
async def internal_api():
112
    return json({"internal": True})
113
```
114

115
## CSRF Protection
116

117
Cross-Site Request Forgery protection for web applications.
118

119
### CSRF Setup
120

121
```python { .api }
122
from blacksheep.server.csrf import CSRFMiddleware, CSRFTokenProvider
123

124
# Basic CSRF protection
125
csrf_middleware = CSRFMiddleware(
126
    secret_key="csrf-secret-key",
127
    cookie_name="csrf_token",
128
    header_name="X-CSRF-Token",
129
    safe_methods=["GET", "HEAD", "OPTIONS", "TRACE"]
130
)
131

132
app.middlewares.append(csrf_middleware)
133

134
# Advanced CSRF configuration
135
csrf_middleware = CSRFMiddleware(
136
    secret_key="csrf-secret-key",
137
    cookie_name="csrf_token",
138
    header_name="X-CSRF-Token",
139
    form_field_name="csrf_token",  # Form field name for token
140
    token_length=32,               # Token length in bytes
141
    max_age=3600,                  # Token expiration in seconds
142
    domain=".example.com",         # Cookie domain
143
    secure=True,                   # Secure cookie flag
144
    same_site="Strict"             # SameSite policy
145
)
146
```
147

148
### CSRF Token Usage
149

150
```python { .api }
151
from blacksheep import Request, Response
152

153
# Generate CSRF token for forms
154
@app.get("/form")
155
async def show_form(request: Request):
156
    csrf_token = request.csrf_token  # Auto-generated token
157
    
158
    form_html = f"""
159
    <form method="post" action="/submit">
160
        <input type="hidden" name="csrf_token" value="{csrf_token}">
161
        <input type="text" name="data" placeholder="Enter data">
162
        <button type="submit">Submit</button>
163
    </form>
164
    """
165
    
166
    return HTMLContent(form_html)
167

168
# CSRF-protected endpoint
169
@app.post("/submit")
170
async def submit_form(request: Request, form_data: FromForm[dict]):
171
    # CSRF validation happens automatically in middleware
172
    data = form_data.value
173
    return json({"submitted": True, "data": data})
174

175
# API endpoint with CSRF token in header
176
@app.post("/api/action")
177
async def api_action(request: Request, data: FromJSON[dict]):
178
    # Client should send X-CSRF-Token header
179
    return json({"action": "completed", "data": data.value})
180
```
181

182
### CSRF Exemptions
183

184
```python { .api }
185
from blacksheep.server.csrf import csrf_exempt
186

187
# Exempt specific endpoints from CSRF protection
188
@app.post("/webhook")
189
@csrf_exempt
190
async def webhook_handler(request: Request):
191
    # Webhooks don't need CSRF protection
192
    payload = await request.json()
193
    return json({"webhook": "processed"})
194

195
# API endpoints with other authentication
196
@app.post("/api/oauth")
197
@csrf_exempt 
198
async def oauth_endpoint(request: Request):
199
    # OAuth endpoints have their own protection
200
    return json({"oauth": "handled"})
201
```
202

203
## Sessions
204

205
Secure session management with encryption and signing.
206

207
### Session Configuration
208

209
```python { .api }
210
from blacksheep.sessions import SessionMiddleware
211
from blacksheep.sessions.crypto import Encryptor, Signer, SessionSerializer
212

213
# Basic session setup
214
app.use_sessions(
215
    secret_key="session-secret-key",
216
    session_cookie="session",
217
    session_max_age=3600  # 1 hour
218
)
219

220
# Advanced session configuration
221
custom_encryptor = Encryptor("encryption-key-32-bytes-long-key")
222
custom_signer = Signer("signing-key")
223
custom_serializer = SessionSerializer(
224
    encryptor=custom_encryptor,
225
    signer=custom_signer
226
)
227

228
session_middleware = SessionMiddleware(
229
    secret_key="master-secret-key",
230
    session_cookie="secure_session",
231
    serializer=custom_serializer,
232
    session_max_age=86400,  # 24 hours
233
    cookie_domain=".example.com",
234
    cookie_secure=True,
235
    cookie_same_site="Lax"
236
)
237

238
app.middlewares.append(session_middleware)
239
```
240

241
### Session Usage
242

243
```python { .api }
244
from blacksheep.sessions import Session
245

246
# Login with session
247
@app.post("/login")
248
async def login(request: Request, credentials: FromJSON[dict]):
249
    username = credentials.value.get("username")
250
    password = credentials.value.get("password")
251
    
252
    # Validate credentials
253
    user = await authenticate_user(username, password)
254
    if not user:
255
        raise Unauthorized("Invalid credentials")
256
    
257
    # Set session data
258
    session: Session = request.session
259
    session["user_id"] = user.id
260
    session["username"] = user.username
261
    session["login_time"] = datetime.utcnow().isoformat()
262
    session["permissions"] = user.permissions
263
    
264
    return json({"message": "Login successful"})
265

266
# Access session data
267
@app.get("/dashboard")
268
async def dashboard(request: Request):
269
    session = request.session
270
    
271
    user_id = session.get("user_id")
272
    if not user_id:
273
        return redirect("/login")
274
    
275
    return json({
276
        "user_id": user_id,
277
        "username": session.get("username"),
278
        "login_time": session.get("login_time")
279
    })
280

281
# Logout
282
@app.post("/logout")
283
async def logout(request: Request):
284
    session = request.session
285
    session.clear()  # Clear all session data
286
    return json({"message": "Logged out successfully"})
287

288
# Partial session update
289
@app.post("/preferences")
290
async def update_preferences(request: Request, prefs: FromJSON[dict]):
291
    session = request.session
292
    
293
    if "user_id" not in session:
294
        raise Unauthorized("Not logged in")
295
    
296
    # Update specific session keys
297
    session["theme"] = prefs.value.get("theme", "light")
298
    session["language"] = prefs.value.get("language", "en")
299
    
300
    return json({"preferences": "updated"})
301
```
302

303
## OpenAPI Documentation
304

305
Automatic API documentation generation with interactive UI.
306

307
### OpenAPI Setup
308

309
```python { .api }
310
from blacksheep.server.openapi.v3 import OpenAPIHandler
311
from blacksheep.server.openapi.ui import ReDocMiddleware, SwaggerUIMiddleware
312

313
app = Application()
314

315
# Configure OpenAPI
316
docs = OpenAPIHandler(
317
    info={
318
        "title": "My API",
319
        "version": "1.0.0",
320
        "description": "A comprehensive API built with BlackSheep"
321
    },
322
    servers=[
323
        {"url": "https://api.example.com", "description": "Production"},
324
        {"url": "https://staging-api.example.com", "description": "Staging"}
325
    ]
326
)
327

328
# Add OpenAPI documentation endpoints
329
docs.bind_app(app)
330

331
# Add Swagger UI
332
app.middlewares.append(SwaggerUIMiddleware(docs, path="/docs"))
333

334
# Add ReDoc UI  
335
app.middlewares.append(ReDocMiddleware(docs, path="/redoc"))
336
```
337

338
### API Documentation
339

340
```python { .api }
341
from typing import Optional, List
342
from dataclasses import dataclass
343
from blacksheep.server.openapi.v3 import doc
344

345
@dataclass
346
class User:
347
    id: int
348
    name: str
349
    email: str
350
    age: Optional[int] = None
351

352
@dataclass
353
class CreateUserRequest:
354
    name: str
355
    email: str
356
    age: Optional[int] = None
357

358
@dataclass
359
class ErrorResponse:
360
    error: str
361
    message: str
362
    code: Optional[int] = None
363

364
# Document endpoints with OpenAPI decorators
365
@app.get("/users")
366
@doc(
367
    summary="List all users",
368
    description="Retrieve a paginated list of all users in the system",
369
    responses={
370
        200: "List of users retrieved successfully",
371
        401: "Authentication required"
372
    },
373
    tags=["Users"]
374
)
375
async def list_users(
376
    page: FromQuery[int] = FromQuery(1),
377
    limit: FromQuery[int] = FromQuery(10)
378
) -> List[User]:
379
    # Implementation here
380
    users = await get_users(page.value, limit.value)
381
    return json([user.__dict__ for user in users])
382

383
@app.post("/users")
384
@doc(
385
    summary="Create new user",
386
    description="Create a new user account with the provided information",
387
    responses={
388
        201: "User created successfully",
389
        400: "Invalid user data",
390
        409: "User already exists"
391
    },
392
    tags=["Users"]
393
)
394
async def create_user(data: FromJSON[CreateUserRequest]) -> User:
395
    user = await create_user_in_db(data.value)
396
    return json(user.__dict__)
397

398
@app.get("/users/{user_id}")
399
@doc(
400
    summary="Get user by ID", 
401
    description="Retrieve a specific user by their unique identifier",
402
    responses={
403
        200: "User found and returned",
404
        404: "User not found"
405
    },
406
    tags=["Users"]
407
)
408
async def get_user(user_id: FromRoute[int]) -> User:
409
    user = await get_user_by_id(user_id.value)
410
    if not user:
411
        raise NotFound("User not found")
412
    return json(user.__dict__)
413
```
414

415
### Security Schemes
416

417
```python { .api }
418
# Define security schemes
419
docs.security_schemes = {
420
    "BearerAuth": {
421
        "type": "http",
422
        "scheme": "bearer",
423
        "bearerFormat": "JWT"
424
    },
425
    "ApiKeyAuth": {
426
        "type": "apiKey",
427
        "in": "header",
428
        "name": "X-API-Key"
429
    },
430
    "OAuth2": {
431
        "type": "oauth2",
432
        "flows": {
433
            "authorizationCode": {
434
                "authorizationUrl": "https://auth.example.com/oauth/authorize",
435
                "tokenUrl": "https://auth.example.com/oauth/token",
436
                "scopes": {
437
                    "read": "Read access",
438
                    "write": "Write access"
439
                }
440
            }
441
        }
442
    }
443
}
444

445
# Apply security to endpoints
446
@app.get("/protected")
447
@doc(
448
    summary="Protected endpoint",
449
    security=[{"BearerAuth": []}],
450
    tags=["Protected"]
451
)
452
@auth()
453
async def protected_endpoint():
454
    return json({"protected": True})
455
```
456

457
## Security Headers
458

459
HTTP security headers for improved application security.
460

461
### HSTS (HTTP Strict Transport Security)
462

463
```python { .api }
464
from blacksheep.server.security.hsts import HSTSMiddleware
465

466
# HSTS middleware
467
hsts_middleware = HSTSMiddleware(
468
    max_age=31536000,      # 1 year in seconds
469
    include_subdomains=True,
470
    preload=True           # Include in HSTS preload list
471
)
472

473
app.middlewares.append(hsts_middleware)
474

475
# Custom HSTS configuration
476
hsts_middleware = HSTSMiddleware(
477
    max_age=86400,         # 1 day
478
    include_subdomains=False,
479
    preload=False,
480
    only_https=True        # Only add header for HTTPS requests
481
)
482
```
483

484
### Security Headers Middleware
485

486
```python { .api }
487
async def security_headers_middleware(request: Request, handler):
488
    response = await handler(request)
489
    
490
    # Add security headers
491
    response.headers.add(b"X-Content-Type-Options", b"nosniff")
492
    response.headers.add(b"X-Frame-Options", b"DENY") 
493
    response.headers.add(b"X-XSS-Protection", b"1; mode=block")
494
    response.headers.add(b"Referrer-Policy", b"strict-origin-when-cross-origin")
495
    
496
    # Content Security Policy
497
    csp = (
498
        "default-src 'self'; "
499
        "script-src 'self' 'unsafe-inline'; "
500
        "style-src 'self' 'unsafe-inline'; "
501
        "img-src 'self' data: https:; "
502
        "font-src 'self'; "
503
        "connect-src 'self'; "
504
        "frame-ancestors 'none';"
505
    )
506
    response.headers.add(b"Content-Security-Policy", csp.encode())
507
    
508
    return response
509

510
app.middlewares.append(security_headers_middleware)
511
```
512

513
## Response Compression
514

515
Compress responses to reduce bandwidth usage.
516

517
### Gzip Compression
518

519
```python { .api }
520
from blacksheep.server.gzip import GzipMiddleware
521

522
# Basic gzip compression
523
gzip_middleware = GzipMiddleware(
524
    minimum_size=1024,     # Only compress responses > 1KB
525
    compression_level=6,    # Compression level (1-9)
526
    content_types={         # MIME types to compress
527
        "text/html",
528
        "text/css", 
529
        "text/javascript",
530
        "application/json",
531
        "application/xml"
532
    }
533
)
534

535
app.middlewares.append(gzip_middleware)
536

537
# Advanced compression configuration
538
gzip_middleware = GzipMiddleware(
539
    minimum_size=500,
540
    compression_level=9,    # Maximum compression
541
    content_types={
542
        "text/*",           # All text types
543
        "application/json",
544
        "application/xml",
545
        "application/javascript"
546
    },
547
    exclude_paths={"/health", "/metrics"}  # Paths to exclude
548
)
549
```
550

551
## Templating
552

553
Template engine integration for server-side rendering.
554

555
### Template Configuration
556

557
```python { .api }
558
from blacksheep import use_templates
559
from jinja2 import Environment, FileSystemLoader
560

561
# Configure Jinja2 templates
562
templates_env = Environment(
563
    loader=FileSystemLoader("templates"),
564
    autoescape=True,
565
    enable_async=True
566
)
567

568
use_templates(app, templates_env)
569

570
# Template rendering endpoint
571
@app.get("/page/{name}")
572
async def render_page(name: str, request: Request):
573
    # Render template with context
574
    template = templates_env.get_template(f"{name}.html")
575
    content = await template.render_async({
576
        "title": f"Page: {name}",
577
        "user": request.identity,
578
        "current_time": datetime.now()
579
    })
580
    
581
    return HTMLContent(content)
582

583
# Template with form
584
@app.get("/contact")
585
async def contact_form():
586
    template = templates_env.get_template("contact.html")
587
    content = await template.render_async({
588
        "csrf_token": generate_csrf_token()
589
    })
590
    return HTMLContent(content)
591
```
592

593
## File Handling
594

595
Advanced file serving and handling capabilities.
596

597
### Dynamic File Serving
598

599
```python { .api }
600
from blacksheep.server.files.dynamic import serve_files_dynamic
601
from blacksheep.server.files import DefaultFileOptions
602

603
# Dynamic file serving with real-time discovery
604
serve_files_dynamic(
605
    app,
606
    source_folder="./uploads",
607
    discovery=True,                # Allow directory listing
608
    cache_time=300,                # Cache for 5 minutes
609
    extensions={".jpg", ".png", ".pdf", ".doc"},
610
    root_path="/uploads",
611
    allow_anonymous=False,         # Require authentication
612
    default_file_options=DefaultFileOptions(
613
        cache_time=3600,
614
        content_disposition_type="attachment"  # Force download
615
    )
616
)
617

618
# Secure file serving
619
@app.get("/secure-files/{filename}")
620
@auth()
621
async def secure_file_download(filename: str, request: Request):
622
    # Validate user has access to file
623
    user_id = request.identity.id
624
    if not await user_can_access_file(user_id, filename):
625
        raise Forbidden("Access denied")
626
    
627
    # Serve file securely
628
    file_path = f"secure_uploads/{filename}"
629
    return file(file_path, content_disposition_type="attachment")
630
```
631

632
### File Upload Handling
633

634
```python { .api }
635
import os
636
from pathlib import Path
637

638
@app.post("/upload")
639
async def upload_files(files: FromFiles, request: Request):
640
    uploaded_files = files.value
641
    results = []
642
    
643
    for file_part in uploaded_files:
644
        # Validate file
645
        if not file_part.file_name:
646
            continue
647
        
648
        filename = file_part.file_name.decode()
649
        content_type = file_part.content_type.decode() if file_part.content_type else "unknown"
650
        
651
        # Security: validate file type
652
        allowed_types = {"image/jpeg", "image/png", "application/pdf"}
653
        if content_type not in allowed_types:
654
            results.append({"filename": filename, "error": "File type not allowed"})
655
            continue
656
        
657
        # Security: sanitize filename
658
        safe_filename = sanitize_filename(filename)
659
        
660
        # Save file
661
        upload_path = Path("uploads") / safe_filename
662
        upload_path.parent.mkdir(exist_ok=True)
663
        
664
        with open(upload_path, "wb") as f:
665
            f.write(file_part.data)
666
        
667
        results.append({
668
            "filename": safe_filename,
669
            "size": len(file_part.data),
670
            "type": content_type,
671
            "uploaded": True
672
        })
673
    
674
    return json({"files": results})
675

676
def sanitize_filename(filename: str) -> str:
677
    """Sanitize filename to prevent directory traversal"""
678
    # Remove path components and dangerous characters
679
    safe_name = os.path.basename(filename)
680
    safe_name = "".join(c for c in safe_name if c.isalnum() or c in "._-")
681
    return safe_name[:100]  # Limit length
682
```
683

684
## Utilities and Helpers
685

686
Various utility functions and helper classes.
687

688
### URL and Path Utilities
689

690
```python { .api }
691
from blacksheep.utils import ensure_bytes, ensure_str, join_fragments
692

693
# String/bytes conversion
694
text = "Hello, World!"
695
bytes_data = ensure_bytes(text)  # b"Hello, World!"
696
text_again = ensure_str(bytes_data)  # "Hello, World!"
697

698
# URL path joining
699
path = join_fragments("api", "v1", "users", "123")  # "api/v1/users/123"
700
path_with_slashes = join_fragments("/api/", "/v1/", "/users/")  # "api/v1/users"
701
```
702

703
### Asyncio Utilities
704

705
```python { .api }
706
from blacksheep.utils.aio import get_running_loop
707
import asyncio
708

709
# Get current event loop
710
loop = get_running_loop()
711

712
# Async context management helpers
713
@asynccontextmanager
714
async def database_context():
715
    db = await connect_database()
716
    try:
717
        yield db
718
    finally:
719
        await db.close()
720

721
# Use context manager
722
async with database_context() as db:
723
    users = await db.query("SELECT * FROM users")
724
```
725

726
### Middleware Utilities
727

728
```python { .api }
729
from blacksheep.middlewares import get_middlewares_chain
730

731
# Create middleware chain
732
async def middleware1(request, handler):
733
    print("Middleware 1 - before")
734
    response = await handler(request)
735
    print("Middleware 1 - after")
736
    return response
737

738
async def middleware2(request, handler):
739
    print("Middleware 2 - before") 
740
    response = await handler(request)
741
    print("Middleware 2 - after")
742
    return response
743

744
async def final_handler(request):
745
    return json({"message": "Final handler"})
746

747
# Chain middlewares
748
middlewares = [middleware1, middleware2]
749
chained_handler = get_middlewares_chain(middlewares, final_handler)
750

751
# Execute chain
752
response = await chained_handler(request)
753
# Output:
754
# Middleware 1 - before
755
# Middleware 2 - before  
756
# Middleware 2 - after
757
# Middleware 1 - after
758
```
759

760
These additional features provide comprehensive functionality for building secure, performant, and well-documented web applications with BlackSheep. The framework's modular design allows you to use only the features you need while maintaining excellent performance and developer experience.