Handle SSL Certificates in aiohttp for Secure Connections
In this tutorial, you’ll learn how to handle SSL certificates in aiohttp to ensure secure connections.
You’ll learn how to implement certificate pinning, handle self-signed certificates, manage certificate revocation checks, and much more.
Implement Certificate Pinning
Certificate pinning adds an extra layer of security by verifying the server’s certificate against a known trusted certificate.
Let’s first generate the certificate chain using openssl:
openssl s_client -showcerts -servername wikipedia.org -connect wikipedia.org:443 </dev/null 2>/dev/null > wikipedia_chain.pem
To implement certificate pinning in aiohttp, you can use the ssl.create_default_context() function and set the cafile parameter.
import ssl
import aiohttp
import asyncio
async def fetch_with_cert_pinning(url, pinned_cert_path):
ssl_context = ssl.create_default_context(cafile=pinned_cert_path)
ssl_context.load_default_certs()
async with aiohttp.ClientSession() as session:
async with session.get(url, ssl=ssl_context) as response:
return await response.text()
async def main():
url = "https://wikipedia.org"
pinned_cert_path = "wikipedia_chain.pem"
try:
result = await fetch_with_cert_pinning(url, pinned_cert_path)
print(result[:100])
except aiohttp.ClientConnectorCertificateError as e:
print(f"Certificate verification failed: {e}")
except aiohttp.ClientError as e:
print(f"Request failed: {e}")
if __name__ == "__main__":
asyncio.run(main())
Output:
<!DOCTYPE html> <html lang="en" class="no-js"> <head> <meta charset="utf-8"> <title>Wikipedia</title
This code creates an SSL context with a pinned certificate and uses it in the aiohttp client session.
The server’s certificate is verified against the pinned certificate.
Handle Self-signed Certificates
When working with self-signed certificates, you need to disable certificate verification while still maintaining encryption.
To handle self-signed certificates, you can create a custom SSL context:
import ssl
import aiohttp
import asyncio
async def fetch_with_self_signed(url):
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
async with aiohttp.ClientSession() as session:
async with session.get(url, ssl=ssl_context) as response:
return await response.text()
async def main():
url = "https://self-signed.badssl.com/"
result = await fetch_with_self_signed(url)
print(result[:100]) # Print first 100 characters of the response
asyncio.run(main())
Output:
<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device
This code creates an SSL context that doesn’t verify the hostname or the certificate so you can connect to servers with self-signed certificates.
Note that you should avoid using this method in production.
Use mutual TLS (mTLS) Authentication
Mutual TLS authentication requires both the client and server to present certificates for authentication.
To implement mTLS in aiohttp, let’s generate self-signed certificates (for testing purposes).
You can use OpenSSL to generate a self-signed certificate and key. Here’s an example command:
openssl req -x509 -newkey rsa:4096 -keyout client_key.pem -out client_cert.pem -days 365 -nodes
Then use the generated files like this:
import ssl
import aiohttp
import asyncio
async def fetch_with_mtls(url, client_cert, client_key):
ssl_context = ssl.create_default_context()
ssl_context.load_cert_chain(client_cert, client_key)
async with aiohttp.ClientSession() as session:
async with session.get(url, ssl=ssl_context) as response:
return await response.text()
async def main():
url = "https://wikipedia.org"
client_cert = "client_cert.pem"
client_key = "client_key.pem"
try:
result = await fetch_with_mtls(url, client_cert, client_key)
print(result[:100])
except aiohttp.ClientError as e:
print(f"mTLS authentication failed: {e}")
asyncio.run(main())
Output:
<!DOCTYPE html> <html lang="en" class="no-js"> <head> <meta charset="utf-8"> <title>Wikipedia</title
This code loads the client certificate and key into the SSL context.
The server will verify the client certificate and the client will verify the server certificate.
Configure SSL/TLS Versions and Cipher Suites
Configuring SSL/TLS protocol versions and cipher suites allows you to enforce stronger security policies.
To configure protocol versions and cipher suites in aiohttp:
import ssl
import aiohttp
import asyncio
async def fetch_with_custom_ssl_config(url):
ssl_context = ssl.create_default_context()
ssl_context.set_ciphers('ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384')
ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2 # Use TLS 1.2 or higher
async with aiohttp.ClientSession() as session:
async with session.get(url, ssl=ssl_context) as response:
return await response.text()
async def main():
url = "https://wikipedia.org"
try:
result = await fetch_with_custom_ssl_config(url)
print(result[:100])
except aiohttp.ClientConnectorCertificateError as e:
print(f"Certificate verification failed: {e}")
except aiohttp.ClientError as e:
print(f"Connection error: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
asyncio.run(main())
Output:
<!DOCTYPE html> <html lang="en" class="no-js"> <head> <meta charset="utf-8"> <title>Wikipedia</title
This code creates an SSL context with specific cipher suites and disables older TLS versions.
It ensures that only strong encryption algorithms are used and that connections use TLS 1.2 or higher.
Create Custom Certificate Stores
Custom certificate stores allow you to manage trusted certificates independently of the system’s default store.
To create and use a custom certificate store in aiohttp:
import ssl
import aiohttp
import asyncio
async def fetch_with_custom_store(url, ca_file):
ssl_context = ssl.create_default_context(cafile=ca_file)
async with aiohttp.ClientSession() as session:
async with session.get(url, ssl=ssl_context) as response:
return await response.text()
async def main():
url = "https://wikipedia.org"
ca_file = "custom_ca_bundle.pem"
try:
result = await fetch_with_custom_store(url, ca_file)
print(result[:100])
except ssl.SSLError as e:
print(f"SSL verification failed: {e}")
asyncio.run(main())
Output:
<!DOCTYPE html> <html lang="en" class="no-js"> <head> <meta charset="utf-8"> <title>Wikipedia</title
This code creates an SSL context using a custom CA bundle file.
It allows you to control which root certificates are trusted, which can be useful in corporate environments or when working with internal CAs.
Regarding the CA bundle used in the above code, you can extract the system trusted CA certificates like this:
>cp /etc/ssl/certs/ca-certificates.crt custom_ca_bundle.pem
Mokhtar is the founder of LikeGeeks.com. He is a seasoned technologist and accomplished author, with expertise in Linux system administration and Python development. Since 2010, Mokhtar has built an impressive career, transitioning from system administration to Python development in 2015. His work spans large corporations to freelance clients around the globe. Alongside his technical work, Mokhtar has authored some insightful books in his field. Known for his innovative solutions, meticulous attention to detail, and high-quality work, Mokhtar continually seeks new challenges within the dynamic field of technology.