OCSP Protocol: Complete Certificate Status Checking Guide for 2026

from dataclasses import dataclass
import hashlib


@dataclass
class OCSPRequest:
    """OCSP request structure."""
    issuer_name_hash: bytes
    issuer_key_hash: bytes
    serial_number: bytes
    nonce: bytes = None


def build_ocsp_request(cert, issuer_cert) -> OCSPRequest:
    """Build OCSP request for a certificate."""
    import hashlib
    
    # Hash of issuer's subject name
    issuer_name_hash = hashlib.sha1(
        issuer_cert.subject.public_bytes()
    ).digest()
    
    # Hash of issuer's public key
    issuer_key_hash = hashlib.sha1(
        issuer_cert.public_key().public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
    ).digest()
    
    # Generate nonce
    import os
    nonce = os.urandom(16)
    
    return OCSPRequest(
        issuer_name_hash=issuer_name_hash,
        issuer_key_hash=issuer_key_hash,
        serial_number=cert.serial_number,
        nonce=nonce
    )


def encode_ocsp_request(request: OCSPRequest) -> bytes:
    """Encode OCSP request in DER format."""
    from pyasn1.codec.der import encoder
    from pyasn1Modules import rfc6960
    
    # Build OCSPRequest structure
    req = rfc6960.OCSPRequest()
    
    # Add request list
    req['requestList'] = []
    
    # This is simplified - real implementation needs full ASN.1 encoding
    return encoder.encode(req)

from enum import Enum
from datetime import datetime


class OCSPResponseStatus(Enum):
    """OCSP response status codes."""
    SUCCESSFUL = 0
    MALFORMED_REQUEST = 1
    INTERNAL_ERROR = 2
    TRY_LATER = 3
    SIG_REQUIRED = 4
    UNAUTHORIZED = 6


class OCSPCertStatus(Enum):
    """Certificate status values."""
    GOOD = 0
    REVOKED = 1
    UNKNOWN = 2


@dataclass
class OCSPResponse:
    """OCSP response structure."""
    response_status: OCSPResponseStatus
    produced_at: datetime
    this_update: datetime
    next_update: datetime = None
    cert_status: OCSPCertStatus = None
    revocation_time: datetime = None
    revocation_reason: int = None
    signature: bytes = None


def parse_ocsp_response(response_data: bytes) -> OCSPResponse:
    """Parse OCSP response."""
    from pyasn1.codec.der import decoder
    
    # Decode response
    response, _ = decoder.decode(response_data, asn1Spec=rfc6960.OCSPResponse())
    
    # Extract fields
    return OCSPResponse(
        response_status=OCSPResponseStatus(int(response['responseStatus'])),
        produced_at=datetime.fromtimestamp(
            int(response['responseBytes']['basicOCSPResponse']['producedAt'])
        ),
        this_update=datetime.fromtimestamp(
            int(response['responseBytes']['basicOCSPResponse']['responses'][0]['thisUpdate'])
        ),
        cert_status=OCSPCertStatus(
            int(response['responseBytes']['basicOCSPResponse']['responses'][0]['certStatus']['good'])
        )
    )

import requests
import base64
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.x509 import ocsp as x509_ocsp
from datetime import datetime, timedelta
from typing import Optional


class OCSPClient:
    """Client for checking certificate status via OCSP."""
    
    def __init__(self, timeout: int = 10):
        self.timeout = timeout
    
    def check_certificate(
        self,
        cert_path: str,
        issuer_path: str,
        ocsp_responder: str
    ) -> dict:
        """Check certificate status via OCSP."""
        
        # Load certificates
        with open(cert_path, 'rb') as f:
            cert = x509.load_pem_x509_certificate(f.read(), default_backend())
        
        with open(issuer_path, 'rb') as f:
            issuer = x509.load_pem_x509_certificate(f.read(), default_backend())
        
        # Build OCSP request
        builder = x509_ocsp.OCSPRequestBuilder()
        builder = builder.add_certificate(
            cert, issuer, hashes.SHA1()
        )
        ocsp_request = builder.build()
        
        # Encode request
        request_der = ocsp_request.public_bytes(serialization.Encoding.DER)
        request_b64 = base64.b64encode(request_der).decode('ascii')
        
        # Send request
        url = f"{ocsp_responder.rstrip('/')}/{cert.serial_number}"
        
        headers = {
            'Content-Type': 'application/ocsp-request',
            'Accept': 'application/ocsp-response'
        }
        
        response = requests.post(
            url,
            data=request_der,
            headers=headers,
            timeout=self.timeout
        )
        
        if response.status_code != 200:
            return {
                'status': 'error',
                'message': f"HTTP {response.status_code}"
            }
        
        # Parse response
        ocsp_response = x509.load_der_ocsp_response(response.content)
        
        # Extract status
        status = ocsp_response.response_status
        cert_status = ocsp_response.certificate_status
        
        result = {
            'status': 'success',
            'response_status': str(status.name),
            'produced_at': str(ocsp_response.produced_at),
            'this_update': str(ocsp_response.this_update),
            'next_update': str(ocsp_response.next_update) if ocsp_response.next_update else None
        }
        
        # Add certificate status
        if cert_status == x509_ocsp.OCSPCertStatus.GOOD:
            result['certificate_status'] = 'good'
        elif cert_status == x509_ocsp.OCSPCertStatus.REVOKED:
            result['certificate_status'] = 'revoked'
            result['revocation_time'] = str(ocsp_response.revocation_time)
            result['revocation_reason'] = str(ocsp_response.revocation_reason) if ocsp_response.revocation_reason else None
        else:
            result['certificate_status'] = 'unknown'
        
        return result


def check_certificate_status(
    cert_path: str,
    issuer_path: str,
    ocsp_responder: str
) -> bool:
    """
    Check if certificate is valid.
    
    Returns True if certificate is valid (not revoked).
    """
    client = OCSPClient()
    result = client.check_certificate(cert_path, issuer_path, ocsp_responder)
    
    if result['status'] != 'success':
        # On error, be conservative - treat as potentially revoked
        return False
    
    return result['certificate_status'] == 'good'

from flask import Flask, request, Response
from datetime import datetime, timedelta
import hashlib


app = Flask(__name__)


# In-memory revocation database (use database in production)
revoked_certs = {
    1234567890: {
        'revoked_at': datetime(2026, 1, 15, 10, 0, 0),
        'reason': 1  # keyCompromise
    }
}


def build_ocsp_response(serial_number: int) -> bytes:
    """Build OCSP response for certificate."""
    
    now = datetime.utcnow()
    
    # Check if revoked
    if serial_number in revoked_certs:
        status = 'revoked'
        revocation_time = revoked_certs[serial_number]['revoked_at']
    else:
        status = 'good'
        revocation_time = None
    
    # Build response (simplified - real implementation needs proper ASN.1)
    response = {
        'responseStatus': 'successful',
        'thisUpdate': now,
        'nextUpdate': now + timedelta(hours=24),
        'certStatus': status,
        'revocationTime': revocation_time
    }
    
    return response


@app.route('/ocsp/<int:serial_number>', methods=['POST', 'GET'])
def ocsp_check(serial_number):
    """OCSP endpoint for checking certificate status."""
    
    # In production, validate request and sign response
    response = build_ocsp_response(serial_number)
    
    return Response(
        str(response),
        mimetype='application/ocsp-response'
    )


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

server {
    listen 443 ssl http2;
    server_name example.com;
    
    ssl_certificate /etc/ssl/certs/example.com.crt;
    ssl_certificate_key /etc/ssl/private/example.com.key;
    
    # Enable OCSP stapling
    ssl_stapling on;
    ssl_stapling_verify on;
    
    # Resolver for OCSP lookup
    ssl_stapling_resolver 8.8.8.8 8.8.4.4 valid=300s;
    ssl_stapling_resolver_timeout 5s;
    
    # Cache OCSP responses
    ssl_stapling_file /var/cache/nginx/ocsp.staple;
    
    # Include root CAs for verification
    ssl_trusted_certificate /etc/ssl/certs/ca-bundle.crt;
}

<VirtualHost *:443>
    ServerName example.com
    
    SSLEngine on
    SSLCertificateFile /etc/ssl/certs/example.com.crt
    SSLCertificateKeyFile /etc/ssl/private/example.com.key
    SSLCertificateChainFile /etc/ssl/certs/ca-bundle.crt
    
    # Enable OCSP stapling
    SSLUseStapling on
    SSLStaplingCache "shmcb:logs/ssl_stapling(32768)"
    SSLStaplingStandardCacheTimeout 3600
    SSLStaplingErrorCacheTimeout 600
    SSLStaplingReturnResponderErrors Off
</VirtualHost>

# Check OCSP stapling with OpenSSL
openssl s_client -connect example.com:443 -status -servername example.com

# Look for "OCSP Response Status: successful"
# and "OCSP Response Data:"

# Check with testssl
testssl --standard --ocsp example.com

# Verify certificate chain and OCSP
openssl verify -CAfile ca-bundle.crt -proxy_url http://proxy:8080 \
    -status_timeout 10 example.com.crt

import ssl
import socket
import OpenSSL
from datetime import datetime, timedelta


def fetch_ocsp_response(cert_path: str, issuer_path: str, ocsp_responder: str) -> bytes:
    """Fetch OCSP response from responder."""
    
    # Load certificates
    cert = OpenSSL.crypto.load_certificate(
        OpenSSL.crypto.FILETYPE_PEM,
        open(cert_path).read()
    )
    issuer = OpenSSL.crypto.load_certificate(
        OpenSSL.crypto.FILETYPE_PEM,
        open(issuer_path).read()
    )
    
    # Create OCSP request
    ocsp_req = OpenSSL.crypto.ocsp_create_request(cert, issuer)
    
    # Send request
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((ocsp_responder, 80))
    
    request = b'POST /ocsp HTTP/1.1\r\n'
    request += b'Host: {}\r\n'.format(ocsp_responder.encode())
    request += b'Content-Type: application/ocsp-request\r\n'
    request += b'Content-Length: {}\r\n\r\n'.format(len(ocsp_req)).encode()
    request += ocsp_req
    
    sock.sendall(request)
    response = sock.recv(4096)
    sock.close()
    
    # Extract OCSP response (simplified)
    return response


def staple_ocsp_response(cert_path: str, issuer_path: str) -> bytes:
    """Get OCSP response for stapling."""
    
    ocsp_responder = "ocsp.example.com"  # Configure your responder
    
    # Fetch fresh OCSP response
    return fetch_ocsp_response(cert_path, issuer_path, ocsp_responder)

import schedule
import time


def monitor_ocsp_responders():
    """Monitor OCSP responder availability."""
    
    responders = [
        'http://ocsp.digicert.com',
        'http://ocsp.sectigo.com',
        'http://ocsp.pki.goog'
    ]
    
    import requests
    
    for responder in responders:
        try:
            start = time.time()
            response = requests.get(responder, timeout=5)
            elapsed = time.time() - start
            
            if response.status_code == 200:
                print(f"✓ {responder}: OK ({elapsed*1000:.0f}ms)")
            else:
                print(f"✗ {responder}: HTTP {response.status_code}")
                
        except Exception as e:
            print(f"✗ {responder}: {e}")


# Check every 5 minutes
schedule.every(5).minutes.do(monitor_ocsp_responders)