基于DRF的Django应用与固件SSL Socket连接实现求助

Hey there! Let's figure out how to switch from your existing DRF REST API to an SSL Socket-based setup for your firmware team. I’ll walk you through the key steps, code examples, and deployment tips to get you up and running.

REST APIs work over HTTP with a request-response cycle—each client call is independent. SSL Sockets, on the other hand, create a persistent, bidirectional connection between your server and the firmware. This is great for real-time or low-latency communication, which is probably why your firmware team wants the switch.

Since you’re already working with Django (Python), we’ll use Python’s built-in socket and ssl modules. First, generate SSL certificates (you can use OpenSSL for this):

openssl req -x509 -newkey rsa:4096 -keyout server.key -out server.crt -days 365 -nodes

This creates a self-signed certificate (fine for testing; use a trusted CA for production).

Next, create your socket server script (e.g., ssl_socket_server.py):

import os
import django
import socket
import ssl
import json

# Initialize Django environment to access ORM
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')
django.setup()

# Import your Django model (adjust to match your app)
from your_app.models import YourDataModel

# Set up SSL context
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(certfile='server.crt', keyfile='server.key')

# Start socket server
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 8443))  # Listen on port 8443
server_socket.listen(5)

print("SSL Socket server running on port 8443...")

while True:
    # Accept incoming connections
    client_conn, client_addr = server_socket.accept()
    with ssl_context.wrap_socket(client_conn, server_side=True) as ssl_conn:
        print(f"New connection from {client_addr}")
        
        # Receive data from firmware (adjust buffer size as needed)
        raw_data = ssl_conn.recv(1024).decode('utf-8')
        if not raw_data:
            ssl_conn.close()
            continue
        
        # Parse request (we'll use JSON for structured commands)
        try:
            request = json.loads(raw_data)
        except json.JSONDecodeError:
            ssl_conn.send(json.dumps({"status": "error", "message": "Invalid JSON"}).encode('utf-8'))
            ssl_conn.close()
            continue
        
        response = {}
        # Handle CRUD operations (focus on your existing create/get logic)
        action = request.get('action')
        if action == 'get':
            # Fetch a record by ID
            item_id = request.get('id')
            try:
                item = YourDataModel.objects.get(id=item_id)
                # Serialize data (match your model fields)
                response = {
                    "status": "success",
                    "data": {
                        "id": item.id,
                        "field1": item.field1,
                        "field2": item.field2
                    }
                }
            except YourDataModel.DoesNotExist:
                response = {"status": "error", "message": "Item not found"}
        
        elif action == 'create':
            # Create a new record
            item_data = request.get('data')
            try:
                new_item = YourDataModel.objects.create(**item_data)
                response = {"status": "success", "data": {"id": new_item.id}}
            except Exception as e:
                response = {"status": "error", "message": str(e)}
        
        # Send response back to firmware
        ssl_conn.send(json.dumps(response).encode('utf-8'))
        ssl_conn.close()

To make sure everything works, write a quick client to simulate the firmware:

import socket
import ssl
import json

# SSL context (disable verification for testing; enable in production)
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    with ssl_context.wrap_socket(sock, server_hostname='your-server-ip') as ssl_sock:
        ssl_sock.connect(('your-server-ip', 8443))
        
        # Test a GET request
        get_request = json.dumps({"action": "get", "id": 1}).encode('utf-8')
        ssl_sock.send(get_request)
        get_response = json.loads(ssl_sock.recv(1024).decode('utf-8'))
        print("GET Response:", get_response)
        
        # Test a CREATE request
        create_request = json.dumps({
            "action": "create",
            "data": {"field1": "test_value", "field2": "another_value"}
        }).encode('utf-8')
        ssl_sock.send(create_request)
        create_response = json.loads(ssl_sock.recv(1024).decode('utf-8'))
        print("CREATE Response:", create_response)

Your existing Gunicorn setup runs the Django WSGI app, but the socket server is a separate process. Here’s how to manage it:

• Systemd Service: Create a service file to keep the socket server running in the background (e.g., /etc/systemd/system/ssl-socket-server.service):

[Unit]
Description=SSL Socket Server for Django App
After=network.target

[Service]
User=your_system_user
WorkingDirectory=/path/to/your/django/project
ExecStart=/usr/bin/python3 /path/to/your/ssl_socket_server.py
Restart=always

[Install]
WantedBy=multi-user.target

Enable and start the service:

sudo systemctl daemon-reload
sudo systemctl start ssl-socket-server
sudo systemctl enable ssl-socket-server

• Nginx Proxy (Optional): If you want to route traffic through Nginx (e.g., use port 443 instead of 8443), use Nginx’s stream module:

stream {
    server {
        listen 443 ssl;
        ssl_certificate /path/to/your/nginx_cert.crt;
        ssl_certificate_key /path/to/your/nginx_cert.key;
        proxy_pass 127.0.0.1:8443;
    }
}

Make sure Nginx has the stream module enabled (most default installations do, but double-check).

• Certificates: Replace self-signed certificates with ones from a trusted CA (like Let’s Encrypt) for production.
• Authentication: Add client-side authentication (e.g., API keys in the request, or client certificate verification) to prevent unauthorized access.
• Concurrency: The basic server handles one connection at a time. For high traffic, use threading or asyncio to handle multiple clients.
• Logging: Add detailed logging to track connections, requests, and errors—this will save you time debugging.

内容的提问来源于stack exchange，提问作者Pratibha