SSE and CORS — Complete Guide
In this tutorial, you will learn about SSE and CORS. We cover key concepts, practical examples, and best practices to help you master this topic.
Handle CORS with Server-Sent Events: configure cross-origin headers, preflight requests, cookie-based authentication, and secure cross-domain SSE streaming in production.
What You Learn
You will learn how CORS works with SSE, how to configure cross-origin headers, handle preflight requests, use cookies for authentication in cross-origin SSE, and secure SSE endpoints.
Why It Matters
Modern web applications often serve the API from a different domain than the frontend. SSE connections from a different origin require proper CORS configuration. Without it, the browser blocks the connection.
Real-World Use
DodaTech's dashboard is served from dashboard.dodatech.com while the API (including SSE) is at api.dodatech.com. CORS headers allow the cross-origin SSE connection. Cookies authenticate the user across both domains.
Basic CORS for SSE
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
class CORSSSEHandler(BaseHTTPRequestHandler):
def do_GET(self):
# Set CORS headers
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(b"event: connected\ndata: {\"status\":\"cors enabled\"}\n\n")
def do_OPTIONS(self):
# Handle preflight requests
self.send_response(200)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Last-Event-ID')
self.send_header('Access-Control-Max-Age', '86400')
self.end_headers()
CORS with Specific Origin
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class SpecificOriginSSE(BaseHTTPRequestHandler):
ALLOWED_ORIGINS = [
'https://dashboard.dodatech.com',
'https://admin.dodatech.com',
]
def get_origin(self):
return self.headers.get('Origin', '')
def is_origin_allowed(self, origin):
return origin in self.ALLOWED_ORIGINS
def do_GET(self):
origin = self.get_origin()
if origin and not self.is_origin_allowed(origin):
self.send_response(403)
self.end_headers()
return
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
if origin:
self.send_header('Access-Control-Allow-Origin', origin)
self.send_header('Vary', 'Origin')
self.end_headers()
self.wfile.write(b"data: {\"status\":\"authenticated\"}\n\n")
def do_OPTIONS(self):
origin = self.get_origin()
if origin and not self.is_origin_allowed(origin):
self.send_response(403)
self.end_headers()
return
self.send_response(200)
if origin:
self.send_header('Access-Control-Allow-Origin', origin)
self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Last-Event-ID')
self.send_header('Access-Control-Allow-Credentials', 'true')
self.send_header('Access-Control-Max-Age', '86400')
self.end_headers()
Cross-Origin Cookies for Auth
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
class CookieAuthSSE(BaseHTTPRequestHandler):
def do_GET(self):
origin = self.headers.get('Origin', '')
cookie = self.headers.get('Cookie', '')
session_token = None
# Parse session token from cookie
if cookie:
for part in cookie.split(';'):
part = part.strip()
if part.startswith('session='):
session_token = part[8:]
if not session_token:
self.send_response(401)
self.end_headers()
return
# Validate session (simplified)
if not self.validate_session(session_token):
self.send_response(403)
self.end_headers()
return
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
if origin:
self.send_header('Access-Control-Allow-Origin', origin)
self.send_header('Access-Control-Allow-Credentials', 'true')
self.send_header('Vary', 'Origin')
self.end_headers()
for i in range(5):
data = json.dumps({'event_id': i, 'user': session_token[:8]})
self.wfile.write(f"data: {data}\n\n".encode())
time.sleep(1)
def validate_session(self, token):
return len(token) > 8
def do_OPTIONS(self):
origin = self.headers.get('Origin', '')
self.send_response(200)
if origin:
self.send_header('Access-Control-Allow-Origin', origin)
self.send_header('Access-Control-Allow-Credentials', 'true')
self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Last-Event-ID, Cookie')
self.send_header('Access-Control-Max-Age', '86400')
self.end_headers()
Client-Side CORS Configuration
// EventSource with CORS and cookies
// The browser automatically handles CORS for EventSource.
// For cookies, the EventSource API does not support credentials option.
// Option 1: Use URL parameters for auth
const token = getAuthToken();
const source = new EventSource(`https://api.dodatech.com/sse/events?token=${token}`);
// Option 2: Use a cookie (set by the server)
// The browser sends cookies automatically if:
// 1. The server sets Access-Control-Allow-Credentials: true
// 2. The server sets Access-Control-Allow-Origin to the exact origin (not "*")
// 3. The cookie is SameSite=None; Secure
// Option 3: Use a fetch-based polyfill for full control
async function createSSEwithHeaders(url, headers) {
const response = await fetch(url, {
headers: {
'Authorization': `Bearer ${token}`,
...headers,
},
});
if (!response.ok) {
throw new Error(`SSE connection failed: ${response.status}`);
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
// Parse SSE format from chunk
processSSEChunk(chunk);
}
}
CORS Middleware Implementation
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class CORSMiddleware:
def __init__(self, allowed_origins=None, allow_credentials=True):
self.allowed_origins = allowed_origins or ['*']
self.allow_credentials = allow_credentials
def get_cors_headers(self, request_origin):
headers = {}
if '*' in self.allowed_origins:
headers['Access-Control-Allow-Origin'] = '*'
elif request_origin in self.allowed_origins:
headers['Access-Control-Allow-Origin'] = request_origin
headers['Vary'] = 'Origin'
if self.allow_credentials:
headers['Access-Control-Allow-Credentials'] = 'true'
else:
return None
return headers
def apply_to_sse(self, handler):
origin = handler.headers.get('Origin', '')
cors_headers = self.get_cors_headers(origin)
if cors_headers is None and origin:
handler.send_response(403)
handler.end_headers()
return False
return cors_headers
cors = CORSMiddleware(
allowed_origins=[
'https://dashboard.dodatech.com',
'https://admin.dodatech.com',
],
allow_credentials=True,
)
class ProtectedSSE(BaseHTTPRequestHandler):
def do_GET(self):
cors_result = cors.apply_to_sse(self)
if cors_result is False:
return
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
if cors_result:
for key, value in cors_result.items():
self.send_header(key, value)
self.end_headers()
self.wfile.write(b"data: {\"message\":\"Authenticated SSE stream\"}\n\n")
Common Mistakes
1. Using Wildcard with Credentials
Access-Control-Allow-Origin: * cannot be used with Access-Control-Allow-Credentials: true. You must specify the exact origin.
2. Missing Vary: Origin
Without Vary: Origin header, CDNs and browsers cache CORS responses incorrectly, causing errors for different origins.
3. Not Handling Preflight
SSE GET requests do not trigger preflight for simple requests. But custom headers or credentials may trigger OPTIONS preflight.
4. CORS Errors Silent in EventSource
EventSource does not provide detailed CORS error messages. Check the browser console for CORS errors.
5. Exposing Authentication Tokens in URLs
Passing tokens as URL parameters (for EventSource) exposes them in server logs and browser history. Use cookies with SameSite=None; Secure instead.
Practice Questions
1. Why does EventSource not support custom headers?
The EventSource API is intentionally simple. For custom headers (Authorization), use a fetch-based Polyfill or pass tokens via cookies/URL.
2. What is the Vary: Origin header used for?
It tells caches that the response varies based on the Origin request header. Different origins get different cached responses.
3. Can you use Access-Control-Allow-Origin: * with credentials?
No. You must specify the exact origin. Using * with Access-Control-Allow-Credentials: true is invalid.
4. How do you authenticate cross-origin SSE connections?
Use cookies with SameSite=None; Secure, or pass an auth token as a URL parameter (less secure but works with EventSource).
Challenge
Configure CORS for SSE in a multi-domain environment: main dashboard (dashboard.dodatech.com), admin panel (admin.dodatech.com), public status page (status.dodatech.com). The status page uses * origin, admin uses credentials. All must handle OPTIONS preflight correctly.
FAQ
Mini Project: CORS-Enabled SSE Server
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
class CORSEnabledSSE(BaseHTTPRequestHandler):
ALLOWED_ORIGINS = ['http://localhost:3000', 'https://dashboard.dodatech.com']
def get_cors_headers(self):
origin = self.headers.get('Origin', '')
headers = {}
if origin in self.ALLOWED_ORIGINS:
headers['Access-Control-Allow-Origin'] = origin
headers['Access-Control-Allow-Credentials'] = 'true'
headers['Vary'] = 'Origin'
elif '*' in self.ALLOWED_ORIGINS:
headers['Access-Control-Allow-Origin'] = '*'
else:
return None, origin
return headers, origin
def do_OPTIONS(self):
cors_headers, origin = self.get_cors_headers()
if cors_headers is None and origin:
self.send_response(403)
else:
self.send_response(200)
if cors_headers:
for k, v in cors_headers.items():
self.send_header(k, v)
self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Last-Event-ID')
self.send_header('Access-Control-Max-Age', '86400')
self.end_headers()
def do_GET(self):
cors_headers, origin = self.get_cors_headers()
if cors_headers is None and origin:
self.send_response(403)
self.end_headers()
return
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
if cors_headers:
for k, v in cors_headers.items():
self.send_header(k, v)
self.end_headers()
for i in range(10):
data = json.dumps({'event': i, 'origin': origin or 'direct'})
self.wfile.write(f"data: {data}\n\n".encode())
time.sleep(1)
server = HTTPServer(('localhost', 8080), CORSEnabledSSE)
print("CORS-enabled SSE on :8080")
server.serve_forever()
What's Next
Now that you understand SSE and CORS, explore SSE in production, then build the mini project: live dashboard.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro