From 220b261263b86274688c3f707a3082590ae298f7 Mon Sep 17 00:00:00 2001 From: Alexander Rossmanith Date: Sun, 16 Aug 2026 16:59:37 +0530 Subject: [PATCH] added aws_lambda code --- aws_lambda/domaincheck/lambda_function.py | 62 +++++++ aws_lambda/lambda-policy.json | 11 ++ aws_lambda/setup-rolesanywhere.ps1 | 99 ++++++++++ aws_lambda/trust-policy.json | 12 ++ networkcheck.code-workspace | 8 + networkcheck/main.py | 208 ++++++++++++++-------- 6 files changed, 326 insertions(+), 74 deletions(-) create mode 100644 aws_lambda/domaincheck/lambda_function.py create mode 100644 aws_lambda/lambda-policy.json create mode 100644 aws_lambda/setup-rolesanywhere.ps1 create mode 100644 aws_lambda/trust-policy.json create mode 100644 networkcheck.code-workspace diff --git a/aws_lambda/domaincheck/lambda_function.py b/aws_lambda/domaincheck/lambda_function.py new file mode 100644 index 0000000..b279bdd --- /dev/null +++ b/aws_lambda/domaincheck/lambda_function.py @@ -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)} diff --git a/aws_lambda/lambda-policy.json b/aws_lambda/lambda-policy.json new file mode 100644 index 0000000..f14278d --- /dev/null +++ b/aws_lambda/lambda-policy.json @@ -0,0 +1,11 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "AllowInvokeDomainCheckLambda", + "Effect": "Allow", + "Action": "lambda:InvokeFunction", + "Resource": "arn:aws:lambda:eu-west-1:458835642486:function:domainCheck" + } + ] +} diff --git a/aws_lambda/setup-rolesanywhere.ps1 b/aws_lambda/setup-rolesanywhere.ps1 new file mode 100644 index 0000000..c77ddf7 --- /dev/null +++ b/aws_lambda/setup-rolesanywhere.ps1 @@ -0,0 +1,99 @@ +# This script sets up AWS Roles Anywhere with a custom CA and an IAM role for Lambda invocation. +# Prerequisites: +# - AWS CLI v2 installed, configured and logged in +# - OpenSSL +# +# https://docs.aws.amazon.com/pdfs/rolesanywhere/latest/userguide/rolesanywhere-guide.pdf +# + +# Variables (replace with your own values) +$AccountId = "458835642486" +$Region = "eu-west-1" +$LambdaArn = "arn:aws:lambda:eu-west-1:$AccountId:function:domainCheck" + +# Paths for certs +$CertDir = Join-Path $PSScriptRoot "rolesanywhere" +New-Item -ItemType Directory -Force -Path $CertDir | Out-Null + +# +$TrustPolicyFile = Join-Path $PSScriptRoot "trust-policy.json" +$LambdaPolicyFile = Join-Path $PSScriptRoot "lambda-policy.json" + +function Convert-ToAwsFileUrl { + param( + [Parameter(Mandatory)] + [string]$Path + ) + + # Expand relative paths and resolve directory separators + $full = (Resolve-Path -LiteralPath $Path).Path + + # Convert backslashes to forward slashes + $normalized = $full -replace '\\','/' + + # Prepend AWS-style file scheme + "file://$normalized" +} + +# Generate CA key + certificate +openssl genrsa -out "$CertDir\ca.key" 2048 +openssl req -x509 -new -nodes -key "$CertDir\ca.key" -sha256 -days 365 ` + -out "$CertDir\ca.pem" -subj "/CN=MyTestCA" ` + -addext "basicConstraints=CA:TRUE" ` + -addext "keyUsage=keyCertSign,cRLSign" + +# Generate client key + certificate signed by CA +openssl genrsa -out "$CertDir\client.key" 2048 +openssl req -new -key "$CertDir\client.key" -out "$CertDir\client.csr" -subj "/CN=MyPC" +openssl x509 -req -in "$CertDir\client.csr" -CA "$CertDir\ca.pem" -CAkey "$CertDir\ca.key" ` + -CAcreateserial -out "$CertDir\client.pem" -days 365 -sha256 + +Write-Host "Certificates generated in $CertDir" + +# Create Trust Anchor +$TrustAnchorArn = aws rolesanywhere create-trust-anchor ` + --region $Region ` + --name MyPCCA ` + --source "sourceData={x509CertificateData=$(Get-Content -Raw $CertDir\ca.pem)},sourceType=CERTIFICATE_BUNDLE" ` + --query 'trustAnchor.trustAnchorArn' --output text + +Write-Host "Trust Anchor ARN: $TrustAnchorArn" + +# Create IAM Role +$TrustPolicUrl = Convert-ToAwsFileUrl -Path $TrustPolicyFile +aws iam create-role ` + --role-name DomainCheckInvokerRole ` + --assume-role-policy-document $TrustPolicUrl + +Write-Host "IAM Role created: DomainCheckInvokerRole" + +# Attach Lambda policy +$LambdaPolicyUrl = Convert-ToAwsFileUrl -Path $LambdaPolicyFile +aws iam put-role-policy ` + --role-name DomainCheckInvokerRole ` + --policy-name LambdaInvokeDomainCheck ` + --policy-document $LambdaPolicyUrl + +Write-Host "Policy attached to role: LambdaInvokeDomainCheck" + +# Create Profile +$ProfileArn = aws rolesanywhere create-profile ` + --region eu-west-1 ` + --name DomainCheckProfile ` + --role-arns '["arn:aws:iam::458835642486:role/DomainCheckInvokerRole"]' ` + --query "profile.profileArn" ` + --output text + +Write-Host "Profile ARN: $ProfileArn" + +# Step 5: Print AWS config block +Write-Host "" +Write-Host "`nPaste the following into your AWS config (~/.aws/config):`n" +Write-Host "[profile rolesanywhere]" +Write-Host "credential_process = `"C:\path\to\aws_signing_helper.exe`" credential-process ^" +Write-Host " --certificate C:\rolesanywhere\client.pem ^" +Write-Host " --private-key C:\rolesanywhere\client.key ^" +Write-Host " --trust-anchor-arn $TrustAnchorArn ^" +Write-Host " --profile-arn $ProfileArn ^" +Write-Host " --role-arn arn:aws:iam::$AccountId:role/$RoleName" +Write-Host "region = $Region" \ No newline at end of file diff --git a/aws_lambda/trust-policy.json b/aws_lambda/trust-policy.json new file mode 100644 index 0000000..29a79a4 --- /dev/null +++ b/aws_lambda/trust-policy.json @@ -0,0 +1,12 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": { + "Service": "rolesanywhere.amazonaws.com" + }, + "Action": "sts:AssumeRole" + } + ] +} diff --git a/networkcheck.code-workspace b/networkcheck.code-workspace new file mode 100644 index 0000000..876a149 --- /dev/null +++ b/networkcheck.code-workspace @@ -0,0 +1,8 @@ +{ + "folders": [ + { + "path": "." + } + ], + "settings": {} +} \ No newline at end of file diff --git a/networkcheck/main.py b/networkcheck/main.py index 726e722..c156c2c 100644 --- a/networkcheck/main.py +++ b/networkcheck/main.py @@ -4,6 +4,7 @@ import socket import boto3 import dns.resolver +import dns.reversename import requests from ping3 import ping @@ -19,7 +20,12 @@ public_nameserver_ipv4 = "8.8.8.8" # Google DNS local_nameserver_ipv4 = "192.168.1.10" # NAS IPv4 DNS -connectivity_test_addresses = ( +reverse_lookup_test_addresses = ([ + "192.168.1.10", # nas + "192.168.1.11", # homeassistant +]) + +connectivity_test_addresses = ([ "192.168.1.1", # Router "192.168.1.10", # NAS "192.168.1.11", # Home Assistant on raspberry pi @@ -31,7 +37,8 @@ connectivity_test_addresses = ( "192.168.1.26", # camera "192.168.1.27", # camera "192.168.1.28" # camera -) + "192.168.1.29" # camera +]) open_ports = ({ "192.168.1.1": [80], # Router @@ -45,55 +52,52 @@ open_ports = ({ "192.168.1.26": [80, 443], # camera "192.168.1.27": [80, 443], # camera "192.168.1.28": [80, 443], # camera + "192.168.1.29": [80, 443], # camera }) -local_dns_test_addresses = ( +local_dns_test_addresses = ([ "google.com", - "homeassistant.local", - "nas.local", "homeassistant.vaishnavi.house", - "nas.vaishnavi.house", - "vaishnavi.house", - "rossmanith.family" -) - -public_dns_test_addresses = ( + "calendar.vaishnavi.house", + "cam.vaishnavi.house", + "contacts.vaishnavi.house", + "drive.vaishnavi.house", + "git.vaishnavi.house", "homeassistant.vaishnavi.house", + "jellyfin.vaishnavi.house", "nas.vaishnavi.house", - "vaishnavi.house", - "rossmanith.family" -) + "photos.vaishnavi.house", + "vpn.vaishnavi.house", + "webdav.vaishnavi.house", +]) -webserver_test_urls = ( - "https://rossmanith.family", - "https://vaishnavi.house", - "https://nas.vaishnavi.house", - "https://homeassistant.vaishnavi.house", -) +public_dns_test_addresses = ([ + "rossmanith.family", + "vaishnavi.house", + "calendar.vaishnavi.house", + "cam.vaishnavi.house", + "contacts.vaishnavi.house", + "drive.vaishnavi.house", + "git.vaishnavi.house", + "homeassistant.vaishnavi.house", + "jellyfin.vaishnavi.house", + "nas.vaishnavi.house", + "photos.vaishnavi.house", + "vpn.vaishnavi.house", + "webdav.vaishnavi.house", +]) + +webserver_test_domains = ([ + "rossmanith.family", + "vaishnavi.house", + "nas.vaishnavi.house", + "homeassistant.vaishnavi.house", +]) def main(): logger.info("Starting network check...") - access_key = "AKIAWVVF4JB3NPKIZGG5" - secret_key = "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" - lambda_client = boto3.client("lambda", region_name="eu-west-1", aws_access_key_id=access_key, aws_secret_access_key=secret_key) - - payload = { - "domains": ["rossmanith.family", "vaishnavi.house"] - } - - response = lambda_client.invoke( - FunctionName="arn:aws:lambda:eu-west-1:458835642486:function:domainCheck", - InvocationType="RequestResponse", - Payload=json.dumps(payload) - ) - - result = json.loads(response["Payload"].read()) - print(json.dumps(result, indent=2)) - - - print() logger.info("Checking IPv4 addresses...") for host, expected_ip in expected_ipv4.items(): @@ -118,22 +122,26 @@ def main(): print() logger.info("Checking DNS resolution of local nameserver %s...", local_nameserver_ipv4) for addr in local_dns_test_addresses: - check_dns(local_nameserver_ipv4, addr) + check_dns_forward_lookup(local_nameserver_ipv4, addr) + + print() + logger.info("Checking DNS reverse lookup of local nameserver %s...", local_nameserver_ipv4) + for ip in reverse_lookup_test_addresses: + check_dns_reverse_lookup(local_nameserver_ipv4, ip) print() logger.info("Checking DNS resolution of public nameserver %s...", public_nameserver_ipv4) for addr in public_dns_test_addresses: - check_dns(public_nameserver_ipv4, addr) + check_dns_forward_lookup(public_nameserver_ipv4, addr) print() logger.info("Checking webserver availability from LAN...") - for url in webserver_test_urls: - check_webserver_lan(url) + for url in webserver_test_domains: + check_webserver_lan("https://" + url) print() logger.info("Checking webserver availability from WAN...") - for url in webserver_test_urls: - check_webserver_wan(url) + check_webserver_wan(webserver_test_domains) def check_ipv4(host, expected_ip): @@ -141,11 +149,11 @@ def check_ipv4(host, expected_ip): try: resolved_ip = socket.gethostbyname(host) if resolved_ip == expected_ip: - logger.info("✅ %s has correct IP: %s", host, resolved_ip) + logger.info("%s has correct IP: %s", host, resolved_ip) else: - logger.warning("❌ %s IP mismatch: %s", host, resolved_ip) + logger.warning("%s IP mismatch: %s", host, resolved_ip) except socket.gaierror as e: - logger.error("❌ %s DNS resolution error: %s", host, e) + logger.error("%s DNS resolution error: %s", host, e) def ping_check(host_ip): @@ -153,60 +161,112 @@ def ping_check(host_ip): try: ping_time = ping(host_ip, timeout=2) if ping_time is not None: - logger.info("✅ %s reachable (ping time: %.2f ms)", host_ip, ping_time * 1000) + logger.info("%s reachable (ping time: %.2f ms)", host_ip, ping_time * 1000) except Exception as e: - logger.error("❌ %s unreachable: %s", host_ip, e) + logger.error("%s unreachable: %s", host_ip, e) -def check_dns(nameserver, addr): + +def check_dns_forward_lookup(nameserver, addr): """Check DNS resolution.""" resolver = dns.resolver.Resolver() resolver.nameservers = [nameserver] try: answer_ = resolver.resolve(addr, "A") - logger.info("✅ DNS %s can resolve %s: %s", nameserver, addr, [r.to_text() for r in answer_]) + logger.info("DNS %s can resolve %s: %s", nameserver, addr, [r.to_text() for r in answer_]) except Exception as e: - logger.error("❌ DNS %s cannot resolve %s: %s", nameserver, addr, e) + logger.error("DNS %s cannot resolve %s: %s", nameserver, addr, e) + + +def check_dns_reverse_lookup(nameserver, ip_address): + # Create a resolver instance + resolver = dns.resolver.Resolver() + resolver.nameservers = [nameserver] # explicitly set the DNS server + # Convert IP to reverse DNS name (e.g., 192.168.1.10 → 10.1.168.192.in-addr.arpa) + rev_name = dns.reversename.from_address(ip_address) + try: + # Query PTR record directly from the specified nameserver + answer = resolver.resolve(rev_name, "PTR") + logger.info("DNS %s can reverse lookup %s: %s", nameserver, ip_address, answer[0]) + except socket.herror as e: + logger.error("DNS %s cannot reverse lookup %s: %s", nameserver, ip_address, e) + def check_webserver_lan(url): try: response = requests.get(url, timeout=5, verify=True) if response.ok: - logger.info("✅ %s responded with %d", url, response.status_code) + logger.info("%s responded with %d", url, response.status_code) else: - logger.warning("⚠️ %s responded with %d", url, response.status_code) + logger.warning("%s responded with %d", url, response.status_code) except requests.exceptions.SSLError as e: - logger.error("❌ SSL certificate error for %s: %s", url, e) + logger.error("SSL certificate error for %s: %s", url, e) except requests.exceptions.ConnectionError: - logger.error("❌ Connection failed: %s not reachable.", url) + logger.error("Connection failed: %s not reachable.", url) except requests.exceptions.Timeout: - logger.error("❌ Request timed out for %s.", url) + logger.error("Request timed out for %s.", url) except Exception as e: - logger.error("❌ Unexpected error for %s: %s", url, e) + logger.error("Unexpected error for %s: %s", url, e) -def check_webserver_wan(domain): - try: - # Use isup.me style check - check_url = f"https://isitup.org/{domain.replace('https://','')}.json" - resp = requests.get(check_url, timeout=10).json() - if resp["status_code"] == 1: - print(f"✅ {domain} is externally reachable") - else: - print(f"❌ {domain} is NOT reachable externally") - except Exception as e: - print(f"⚠️ Error checking {domain}: {e}") - +def check_webserver_wan(webserver_test_domains): + """ + Check webserver availability from WAN using AWS Lambda. + This function invokes a Lambda function that checks if the specified hosts are reachable and retrieves SSL information + All hosts are checked in a single Lambda invocation to minimize costs. + """ + access_key = "AKIAWVVF4JB3NPKIZGG5" + secret_key = "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" + lambda_client = boto3.client("lambda", region_name="eu-west-1", aws_access_key_id=access_key, aws_secret_access_key=secret_key) + payload = { + "hostnames": webserver_test_domains + } + response = lambda_client.invoke( + FunctionName="arn:aws:lambda:eu-west-1:458835642486:function:domainCheck", + InvocationType="RequestResponse", + Payload=json.dumps(payload) + ) + Response = json.loads(response["Payload"].read()) + if Response.get("errorType"): + logger.error("Error occurred while checking domains: %s", Response.get("errorMessage")) + return + for hostname in webserver_test_domains: + host_response = Response.get(hostname) + https_get_request_result = host_response.get("https_get_request", {}) + if not https_get_request_result.get("reachable", False): + logger.error("Host %s is not reachable: %s", hostname, https_get_request_result.get("response", "Unknown error")) + continue + logger.info( + "Host %s is reachable. Status code: %s", + hostname, https_get_request_result["status_code"]) + cert_info = host_response.get("ssl", {}) + if not cert_info: + logger.error("Host %s has no SSL certificate information.", hostname) + continue + if not cert_info["valid"]: + logger.error( + "Host %s has an invalid SSL certificate. Issuer: %s. Expiry: %s. Valid: %s", + hostname, + cert_info["issuer"], + cert_info["expiry"], + cert_info["valid"]) + continue + logger.info( + "Host %s has a valid SSL certificate. Issuer: %s, expiry: %s, valid: %s", + hostname, + cert_info["issuer"], + cert_info["expiry"], + cert_info["valid"]) def check_port(addr, port): """Check if a specific port is open on the host.""" try: with socket.create_connection((addr, port), timeout=2): - logger.info("✅ %s:%d is open", addr, port) + logger.info("%s:%d is open", addr, port) except (socket.timeout, ConnectionRefusedError): - logger.warning("⚠️ %s:%d is closed or filtered", addr, port) + logger.warning("%s:%d is closed or filtered", addr, port) except Exception as e: - logger.error("❌ Error checking %s:%d: %s", addr, port, e) + logger.error("Error checking %s:%d: %s", addr, port, e) if __name__ == "__main__":