Automating AWS WAF IP Threat list with ProxyCheck.io

AWSTemplateFormatVersion: '2010-09-09' Transform: AWS::Serverless-2016-10-31 Description: > ProxyCheck WAF automation ProxyCheck WAF automation Lambda Function that reads IPs from S3, checks risk scores using ProxyCheck.io, and updates AWS WAF IP sets with high-risk IPs. # More info about Globals: https://github.com/awslabs/serverless-application-model/blob/master/docs/globals.rst Globals: Function: Timeout: 900 # 15 minutes MemorySize: 512 LoggingConfig: LogFormat: JSON Resources: ProxyCheckWAFAutomation: Type: AWS::Serverless::Function Properties: FunctionName: "ProxyCheckWAFAutomation" CodeUri: proxy_check_waf_automation/ Handler: app.lambda_handler Runtime: python3.13 Architectures: - x86_64 Timeout: 900 # 15 minutes MemorySize: 512 Environment: Variables: S3_BUCKET_NAME: "proxycheck-waf-automation" PROXYCHECK_API_KEY: "xxxx-xxxxx-xxxxx-xxxxx" TELEGRAM_BOT_TOKEN: "YOUR_TG_BOT_TOKEN" TELEGRAM_CHAT_ID: "-xxxxx" REDIS_URL: "redis://username:password@hostname/db_name" Events: Trigger: Type: EventBridgeRule Properties: Pattern: source: - "aws.s3" detail-type: - "Object Created" - "Object Updated" detail: bucket: name: - "proxycheck-waf-automation" Policies: - S3ReadPolicy: BucketName: "proxycheck-waf-automation/*" - Statement: - Effect: Allow Action: - wafv2:GetIPSet - wafv2:UpdateIPSet Resource: "*" 

import json import csv import os import io import logging from typing import List, Set, Optional import boto3 import requests import proxycheck import botocore.exceptions import sys import redis import time # Set up logging logger = logging.getLogger() logger.setLevel(logging.INFO) # Initialize AWS clients s3_client = boto3.client('s3') wafv2_client = boto3.client('wafv2', region_name='us-east-1') # Initialize ProxyCheck client proxy_checker = proxycheck.Blocking(key=os.environ.get('PROXYCHECK_API_KEY')) # Initialize Redis client REDIS_URL = os.environ.get('REDIS_URL') redis_client = None def get_redis_client(): """Get Redis client with connection retry logic""" global redis_client if redis_client is None: try: redis_client = redis.from_url( REDIS_URL, socket_timeout=5, socket_connect_timeout=5, retry_on_timeout=True, health_check_interval=30 ) # Test connection redis_client.ping() logger.info("Redis connection established successfully") except Exception as e: logger.warning(f"Failed to connect to Redis: {str(e)}. Will proceed without caching.") redis_client = None return redis_client def lambda_handler(event, context): """Daily IP Checker Lambda Function This function: 1. Reads a CSV file from S3 containing IPs 2. Deduplicates the IPs 3. Checks risk scores using ProxyCheck.io (limited to 1000 requests) 4. Adds high-risk IPs (score > 33) to AWS WAF IP set 5. Sends Telegram notification about the update """ try: # Get configuration from environment variables s3_bucket = os.environ.get('S3_BUCKET_NAME') s3_key = event["detail"]["object"]["key"] waf_scope = 'CLOUDFRONT' waf_ip_set_name = "ProxyCheckAutomation" waf_ip_set_id = "your_id_here" telegram_bot_token = os.environ.get('TELEGRAM_BOT_TOKEN') telegram_chat_id = os.environ.get('TELEGRAM_CHAT_ID') # Validate required environment variables required_vars = [s3_bucket, s3_key, waf_ip_set_name, waf_ip_set_id] if not all(required_vars): raise ValueError("Missing required environment variables") logger.info(f"Starting daily IP check process") # Step 1: Read CSV file from S3 logger.info(f"Reading CSV file from S3: s3://{s3_bucket}/{s3_key}") ips = read_ips_from_s3(s3_bucket, s3_key) logger.info(f"Found {len(ips)} unique IPs after deduplication") # Step 2: Check IP risk scores (limited to 1000 requests) logger.info("Checking IP risk scores with ProxyCheck.io and Redis caching") risky_ips = check_ip_risk_scores(ips[:1000]) logger.info(f"Found {len(risky_ips)} risky IPs (risk score > 33)") # Step 3: Add risky IPs to AWS WAF IP set added_count = 0 if risky_ips: logger.info("Adding risky IPs to AWS WAF IP set") # Convert IPs to CIDR format networks = {f"{ip}/32" for ip in risky_ips} added_count = update_single_ipset(wafv2_client, waf_ip_set_name, waf_scope, waf_ip_set_id, networks) logger.info(f"Added {added_count} IPs to WAF IP set") else: logger.info("No risky IPs found, no updates needed") # Step 4: Send Telegram notification with cache stats if telegram_bot_token and telegram_chat_id: # Get cache efficiency stats for notification cache_stats = get_cache_stats() send_telegram_notification( telegram_bot_token, telegram_chat_id, len(ips), len(risky_ips), added_count, context, cache_stats ) return { "statusCode": 200, "body": json.dumps({ "message": "Daily IP check completed successfully", "total_ips_processed": len(ips), "risky_ips_found": len(risky_ips), "ips_added_to_waf": added_count if risky_ips else 0 }) } except Exception as e: logger.error(f"Error in daily IP check: {str(e)}") # Send error notification via Telegram if configured telegram_bot_token = os.environ.get('TELEGRAM_BOT_TOKEN') telegram_chat_id = os.environ.get('TELEGRAM_CHAT_ID') if telegram_bot_token and telegram_chat_id: send_error_notification(telegram_bot_token, telegram_chat_id, str(e), context) raise e def read_ips_from_s3(bucket: str, key: str) -> List[str]: """Read and deduplicate IPs from CSV file in S3""" try: # Download the CSV file from S3 response = s3_client.get_object(Bucket=bucket, Key=key) csv_content = response['Body'].read().decode('utf-8') # Parse CSV and extract IPs ips: Set[str] = set() csv_reader = csv.reader(io.StringIO(csv_content)) for row in csv_reader: # Skip empty rows if not row or len(row) < 2: continue # IP is in the second column (index 1) ip_value = row[1].strip() if ip_value and is_valid_ip(ip_value): ips.add(ip_value) return list(ips) except Exception as e: logger.error(f"Error reading IPs from S3: {str(e)}") raise def is_valid_ip(ip: str) -> bool: """Basic IP validation""" try: parts = ip.split('.') return len(parts) == 4 and all(0 <= int(part) <= 255 for part in parts) except (ValueError, AttributeError): return False def get_cached_risk_score(ip: str) -> Optional[int]: """Get cached risk score from Redis""" try: redis_conn = get_redis_client() if redis_conn is None: return None cache_key = f"ip_risk:{ip}" cached_score = redis_conn.get(cache_key) if cached_score is not None: return int(cached_score.decode('utf-8')) return None except Exception as e: logger.warning(f"Error reading from Redis cache for IP {ip}: {str(e)}") return None def cache_risk_score(ip: str, risk_score: int, ttl_hours: int = 720): """Cache risk score in Redis with TTL""" try: redis_conn = get_redis_client() if redis_conn is None: return cache_key = f"ip_risk:{ip}" redis_conn.setex(cache_key, ttl_hours * 3600, str(risk_score)) logger.debug(f"Cached risk score for IP {ip}: {risk_score}") except Exception as e: logger.warning(f"Error caching risk score for IP {ip}: {str(e)}") # Global cache stats for tracking cache_stats = {"cache_hits": 0, "api_calls": 0, "checked_count": 0} def get_cache_stats(): """Get current cache statistics""" return cache_stats.copy() def reset_cache_stats(): """Reset cache statistics""" global cache_stats cache_stats = {"cache_hits": 0, "api_calls": 0, "checked_count": 0} def check_ip_risk_scores(ips: List[str]) -> List[str]: """Check IP risk scores using ProxyCheck.io with Redis caching and return risky IPs""" risky_ips = [] reset_cache_stats() # Reset stats at the beginning of each run try: logger.info(f"Starting to check {len(ips)} IPs with Redis caching") for ip in ips: try: # First, check if we have a cached result cached_risk = get_cached_risk_score(ip) if cached_risk is not None: risk_score = cached_risk cache_stats["cache_hits"] += 1 logger.debug(f"Cache hit for IP {ip}: risk score {risk_score}") else: # Check individual IP using proxy_checker.ip() method ip_result = proxy_checker.ip(ip) risk_score = ip_result.risk() cache_stats["api_calls"] += 1 # Cache the result for future use cache_risk_score(ip, risk_score) logger.debug(f"API call for IP {ip}: risk score {risk_score}") cache_stats["checked_count"] += 1 if risk_score > 33: risky_ips.append(ip) logger.info(f"Risky IP found: {ip} (risk score: {risk_score})") # Log progress every 100 IPs if cache_stats["checked_count"] % 100 == 0: logger.info(f"Progress: {cache_stats['checked_count']}/{len(ips)} IPs processed, {len(risky_ips)} risky IPs found. Cache hits: {cache_stats['cache_hits']}, API calls: {cache_stats['api_calls']}") except Exception as e: logger.warning(f"Error checking IP {ip}: {str(e)}") continue except Exception as e: logger.error(f"Error checking IP risk scores: {str(e)}") raise cache_efficiency = (cache_stats["cache_hits"] / cache_stats["checked_count"] * 100) if cache_stats["checked_count"] > 0 else 0 logger.info(f"Completed checking {cache_stats['checked_count']} IPs, found {len(risky_ips)} risky IPs. Cache efficiency: {cache_stats['cache_hits']}/{cache_stats['checked_count']} ({cache_efficiency:.1f}% cache hits), API calls saved: {cache_stats['cache_hits']}") return risky_ips def get_ipset_and_token(client, name, scope, ipset_id): """ Retrieve an existing WAFv2 IPSet. Returns (lock_token, description, existing_addresses_set). """ try: resp = client.get_ip_set(Name=name, Scope=scope, Id=ipset_id) except botocore.exceptions.ClientError as e: logger.error(f"Error fetching IPSet {name}: {e}") raise ipset = resp["IPSet"] lock_token = resp["LockToken"] description = ipset.get("Description", "") existing = set(ipset.get("Addresses", [])) return lock_token, description, existing def update_single_ipset(client, name, scope, ipset_id, networks, is_ipv6=False): """ Add any networks in `networks` to the specified IPSet. `networks` is a set of CIDR network strings (e.g., '192.168.1.1/32'). Returns the number of new IPs added. """ try: # 1) Fetch existing lock_token, description, existing_addrs = get_ipset_and_token( client, name, scope, ipset_id ) # 2) Compute new to_add = networks - existing_addrs if not to_add: logger.info(f"[{name}] no new {'IPv6' if is_ipv6 else 'IPv4'} CIDRs to add.") return 0 # 3) Merge and sort updated = sorted(existing_addrs | networks) limit = 10000 if len(updated) > limit: logger.error(f"[{name}] total addresses {len(updated)} exceed limit {limit}.") raise ValueError(f"IPSet would exceed limit of {limit} addresses") # 4) Update IPSet try: resp = client.update_ip_set( Name=name, Scope=scope, Id=ipset_id, Description=description or 'Daily IP Risk Checker - High Risk IPs', Addresses=list(updated), LockToken=lock_token ) except botocore.exceptions.ClientError as e: logger.error(f"Error updating IPSet {name}: {e}") raise next_token = resp.get("NextLockToken") logger.info( f"[{name}] Inserted {len(to_add)} new items; now has {len(updated)} addresses. " f"NextLockToken: {next_token}" ) return len(to_add) except Exception as e: logger.error(f"Error updating IPSet {name}: {str(e)}") raise def send_telegram_notification(bot_token: str, chat_id: str, total_ips: int, risky_ips: int, added_ips: int, context=None, cache_stats=None): """Send Telegram notification about the IP check results""" try: request_id = context.aws_request_id if context else 'Unknown' # Build cache efficiency message cache_message = "" if cache_stats and cache_stats.get("checked_count", 0) > 0: cache_efficiency = (cache_stats["cache_hits"] / cache_stats["checked_count"] * 100) cache_message = f""" 💾 Cache Performance: • Cache hits: {cache_stats["cache_hits"]} • API calls: {cache_stats["api_calls"]} • Efficiency: {cache_efficiency:.1f}% • API calls saved: {cache_stats["cache_hits"]}""" message = f"""🛡️ Daily IP Check Report 📊 Total IPs processed: {total_ips} ⚠️ Risky IPs found: {risky_ips} 🔒 IPs added to WAF: {added_ips}{cache_message} Status: {'✅ Completed successfully' if risky_ips >= 0 else '❌ Failed'} Request ID: {request_id} """ url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = { 'chat_id': chat_id, 'text': message, 'parse_mode': 'HTML' } response = requests.post(url, json=payload, timeout=10) response.raise_for_status() logger.info("Telegram notification sent successfully") except Exception as e: logger.error(f"Error sending Telegram notification: {str(e)}") def send_error_notification(bot_token: str, chat_id: str, error_message: str, context=None): """Send Telegram notification about errors""" try: request_id = context.aws_request_id if context else 'Unknown' message = f"""❌ Daily IP Check Error Error: {error_message} Request ID: {request_id} Please check CloudWatch logs for more details. """ url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = { 'chat_id': chat_id, 'text': message, 'parse_mode': 'HTML' } response = requests.post(url, json=payload, timeout=10) response.raise_for_status() except Exception as e: logger.error(f"Error sending error notification: {str(e)}")