added aws_lambda code
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
"""
|
||||
This module contains a Lambda function that checks if domains are reachable and retrieves SSL information.
|
||||
Amazon will bill for the wait time of this function, timeout is set to 10 seconds.
|
||||
TODO: Implement a loop with asynchronous checking of multiple domains.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import ssl
|
||||
import socket
|
||||
from datetime import datetime
|
||||
|
||||
def lambda_handler(event, context):
|
||||
"""
|
||||
Args:
|
||||
event (dict):
|
||||
The event data passed to the Lambda function. Expects a dictionary
|
||||
with a "hostnames" key containing a list of all hosts to be checked.
|
||||
context (LambdaContext):
|
||||
The runtime information of the Lambda function.
|
||||
"""
|
||||
max_timeout = 10
|
||||
remaining_time_in_seconds = context.get_remaining_time_in_millis() / 1000
|
||||
timeout_in_seconds = min(remaining_time_in_seconds - 1, max_timeout)
|
||||
results = {}
|
||||
for hostname in event["hostnames"]:
|
||||
url = f"https://{hostname}/"
|
||||
host_result = {}
|
||||
host_result["https_get_request"] = get_request(url, timeout=timeout_in_seconds)
|
||||
if host_result["https_get_request"].get("reachable"):
|
||||
host_result["ssl"] = ssl_info(hostname)
|
||||
results[hostname] = host_result
|
||||
return results
|
||||
|
||||
|
||||
def get_request(url, timeout=10):
|
||||
result = {}
|
||||
try:
|
||||
resp = requests.get(url, timeout=timeout)
|
||||
result["status_code"] = resp.status_code
|
||||
result["reason"] = resp.reason
|
||||
result["reachable"] = resp.status_code == 200
|
||||
except requests.exceptions.RequestException as e:
|
||||
result["error"] = str(e)
|
||||
result["reachable"] = False
|
||||
return result
|
||||
|
||||
|
||||
def ssl_info(hostname, port=443):
|
||||
try:
|
||||
ctx = ssl.create_default_context()
|
||||
with socket.create_connection((hostname, port), timeout=10) as sock:
|
||||
with ctx.wrap_socket(sock, server_hostname=hostname) as ssock:
|
||||
cert = ssock.getpeercert()
|
||||
issuer = dict(x[0] for x in cert['issuer'])
|
||||
not_after = datetime.strptime(cert['notAfter'], "%b %d %H:%M:%S %Y %Z")
|
||||
return {
|
||||
"issuer": issuer.get("organizationName", "Unknown"),
|
||||
"expiry": not_after.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"valid": not_after > datetime.utcnow()
|
||||
}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
Reference in New Issue
Block a user