Index: /branches/amp_4_0/platform/config/init_db.sql =================================================================== --- /branches/amp_4_0/platform/config/init_db.sql (revision 2630) +++ /branches/amp_4_0/platform/config/init_db.sql (working copy) @@ -113,7 +113,7 @@ connection varchar(16) NOT NULL DEFAULT 'unconnected', status varchar(16) NOT NULL DEFAULT 'new', version varchar(8192) DEFAULT NULL, - license_key varchar(256) DEFAULT NULL, + license_key varchar(128) DEFAULT NULL, gateway_domain varchar(128) DEFAULT NULL, location varchar(128) NOT NULL, firewall_ip varchar(64) NOT NULL, @@ -364,4 +364,19 @@ time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ); +-- Create a adc_vs_ssl_info table +CREATE TABLE IF NOT EXISTS adc_vs_ssl_info ( + id BIGSERIAL PRIMARY KEY, + device_id VARCHAR(64), + device_name VARCHAR(32) NOT NULL, + vs_name VARCHAR(512) NOT NULL, + vhost_name VARCHAR(512) NOT NULL, + cert_type VARCHAR(16) NOT NULL, + expiration INT, + CONSTRAINT unique_adc_vs_ssl_cert UNIQUE (device_id, device_name, vs_name, vhost_name, cert_type), + CONSTRAINT fk_device_id FOREIGN KEY (device_id) + REFERENCES device (id) MATCH SIMPLE + ON UPDATE NO ACTION + ON DELETE NO ACTION +); -- psql -U amp_admin -d cm -f /path/to/your/init_db.sql Index: /branches/amp_4_0/scripts/check_adc_ssl.py =================================================================== --- /branches/amp_4_0/scripts/check_adc_ssl.py (revision 2630) +++ /branches/amp_4_0/scripts/check_adc_ssl.py (working copy) @@ -1,29 +1,113 @@ -#!/usr/bin/python +#!/usr/bin/python3 +import base64 +import datetime import json +import logging +import os import sys -from StringIO import StringIO +from io import StringIO import psycopg2 import pycurl -import requests +import toml +from influxdb_client import InfluxDBClient, Point +from influxdb_client.client.write_api import SYNCHRONOUS +# --- Logging Setup --- +LOG_FILE = '/var/log/check_devices_ssl_certificate.log' +LOG_LEVEL = logging.INFO # Set to logging.DEBUG for more verbose output + +# Ensure the log directory exists and is writable (best practice for production) +log_dir = os.path.dirname(LOG_FILE) +if not os.path.exists(log_dir): + try: + os.makedirs(log_dir, exist_ok=True) + except OSError as e: + logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') + logging.error(f"Failed to create log directory {log_dir}: {e}. Logging to stderr.") + sys.exit(1) # Exit if logging setup is critical + +try: + logging.basicConfig( + filename=LOG_FILE, + level=LOG_LEVEL, + format='%(asctime)s - %(levelname)s - %(message)s' + ) +except Exception as e: + logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') + logging.error(f"Failed to set up file logging to {LOG_FILE}: {e}. Logging to stderr.") + # If logging to the file fails, we can choose to continue to stderr or exit. + # For a critical script, exiting might be safer. + sys.exit(1) + +logger = logging.getLogger(__name__) + sys.path.append("/ca/webui/htdocs/new/src") -from djproject.an_settings import * +try: + from djproject.an_settings import * +except ImportError as e: + logger.critical( + f"Failed to import djproject.an_settings: {e}. Please ensure the path is correct and the module exists.") + sys.exit(1) -READ_QUERY_URL = 'http://127.0.0.1:8086/query?db=composer&epoch=ms' -WRITE_QUERY_URL = 'http://127.0.0.1:8086/write?db=composer&epoch=ms' +# --- InfluxDB 2.x Configuration --- +# IMPORTANT: Load these from environment variables in production for security. +INFLUXDB_URL = os.getenv("INFLUXDB_V2_URL", "http://localhost:8086") +INFLUXDB_TOKEN = os.getenv("INFLUXDB_V2_TOKEN", "YOUR_INFLUXDB_V2_TOKEN") +# --- NEW: Read InfluxDB Token from a TOML file --- +INFLUXDB_TOKEN_FILE = "/opt/influxdb2_token.toml" +try: + with open(INFLUXDB_TOKEN_FILE, 'r') as f: + config = toml.load(f) + INFLUXDB_TOKEN = config['influxdb']['token'] + logger.info(f"Successfully loaded InfluxDB token from {INFLUXDB_TOKEN_FILE}") +except FileNotFoundError: + logger.critical(f"InfluxDB token file not found: {INFLUXDB_TOKEN_FILE}. Exiting.") + sys.exit(1) +except KeyError as e: + logger.critical( + f"Key '{e}' not found in InfluxDB token file {INFLUXDB_TOKEN_FILE}. Ensure '[influxdb]' and 'token = ...' exist. Exiting.") + sys.exit(1) +except Exception as e: + logger.critical(f"Failed to read InfluxDB token from {INFLUXDB_TOKEN_FILE}: {e}. Exiting.") + sys.exit(1) +# Ensure the token was loaded +if not INFLUXDB_TOKEN: + logger.critical("InfluxDB token is empty after attempting to load from file. Exiting.") + sys.exit(1) + +INFLUXDB_ORG = os.getenv("INFLUXDB_V2_ORG", "AN") # <<< REPLACE THIS or set env var +INFLUXDB_BUCKET_CERTS_METADATA = os.getenv("INFLUXDB_V2_BUCKET_CERTS_METADATA", "AMP") # Bucket for cert metadata +INFLUXDB_BUCKET_EVENTS = os.getenv("INFLUXDB_V2_BUCKET_EVENTS", "event_detection") # Bucket for event data + +# Initialize InfluxDB 2.x client +try: + influx_client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG) + influx_write_api = influx_client.write_api(write_options=SYNCHRONOUS) + influx_query_api = influx_client.query_api() + logger.info("Successfully initialized InfluxDB 2.x client.") +except Exception as e: + logger.critical(f"Failed to initialize InfluxDB 2.x client: {e}. Exiting.") + sys.exit(1) + + def modify_url(url, device_type, vsite_name=None): + """ + Dynamically modifies a base URL based on device type and optional vsite name. + """ _type = None new_url = None - # query device type from string - # like: 'Array APV 3600 v5' -> 'APV' + + # Query device type from string like: 'Array APV 3600 v5' -> 'APV' for each in device_type.split(): if each.lower() in DEVICE_TYPE_LIST: _type = each.lower() break + if not _type: + logger.warning(f"Unknown device type '{device_type}'. Cannot modify URL.") return new_url url_list = url.split('/') @@ -43,14 +127,22 @@ url_list[2] = 'apv' elif _type in OTHER_ADC_LIST: url_list[2] = _type if _type[0] != "v" else _type[1:] + new_url = '/'.join(url_list) + logger.debug(f"Modified URL for device type '{device_type}': {new_url}") return str(new_url) def call_restapi_multi(option_list): + """ + Makes concurrent HTTP/HTTPS API calls using pycurl.CurlMulti. + """ multi = pycurl.CurlMulti() handles = [] result = [] + + logger.info(f"Initiating {len(option_list)} concurrent REST API calls.") + for options in option_list: method = options[0] url = options[1] @@ -61,217 +153,426 @@ restapi_password = options[6] proto = options[7] cookie = options[8] - authorization = 'Basic %s' % ( - ('%s:%s' % (restapi_username, restapi_password)).encode('base64').replace('\n', '')) + + auth_string = f"{restapi_username}:{restapi_password}".encode('utf-8') + authorization = 'Basic %s' % base64.b64encode(auth_string).decode('utf-8') + curl = pycurl.Curl() curl.body = StringIO() - curl.setopt(pycurl.VERBOSE, 0) # DEBUG MODE + + curl.setopt(pycurl.VERBOSE, 0) curl.setopt(pycurl.CONNECTTIMEOUT, 5) - curl.setopt(pycurl.HTTPHEADER, ['Accept: application/json', 'Authorization: %s' % authorization, 'Expect: ']) + curl.setopt(pycurl.HTTPHEADER, ['Accept: application/json', f'Authorization: {authorization}', 'Expect: ']) + if method == 'POST': curl.setopt(pycurl.POST, 1) - curl.setopt(pycurl.POSTFIELDS, params) - if method == 'PUT': + curl.setopt(pycurl.POSTFIELDS, params.encode('utf-8')) + elif method == 'PUT': curl.setopt(pycurl.CUSTOMREQUEST, "PUT") - curl.setopt(pycurl.POSTFIELDS, params) - curl.setopt(pycurl.HEADER, 0) # don't write http header to response + curl.setopt(pycurl.POSTFIELDS, params.encode('utf-8')) + + curl.setopt(pycurl.HEADER, 0) + if proto == 'https': - curl.setopt(pycurl.SSL_VERIFYPEER, 0) - curl.setopt(pycurl.SSL_VERIFYHOST, 0) - curl.setopt(pycurl.URL, str("https://%s:%d%s" % (ip, restapi_port, url))) + # --- SECURITY WARNING --- + # Disabling SSL verification (SSL_VERIFYPEER, SSL_VERIFYHOST) is a SECURITY RISK. + # It makes your connection vulnerable to Man-in-the-Middle attacks. + # In production, set SSL_VERIFYPEER to 1 and SSL_CACERT to a trusted CA bundle. + # If your devices use self-signed certificates, import their CAs into a trust store + # and point SSL_CACERT to it. + curl.setopt(pycurl.SSL_VERIFYPEER, 0) # Set to 1 in production and configure SSL_CACERT + curl.setopt(pycurl.SSL_VERIFYHOST, 0) # Set to 2 in production + curl.setopt(pycurl.URL, f"https://{ip}:{restapi_port}{url}") else: - curl.setopt(pycurl.URL, str("http://%s:%d%s" % (ip, restapi_port, url))) + curl.setopt(pycurl.URL, f"http://{ip}:{restapi_port}{url}") + curl.setopt(pycurl.WRITEFUNCTION, curl.body.write) if cookie: curl.setopt(pycurl.COOKIE, cookie) + multi.add_handle(curl) handles.append(curl) num_handles = len(option_list) while num_handles: - ret, num_handles = multi.perform() - if ret != pycurl.E_CALL_MULTI_PERFORM: - if num_handles: - multi.select(0.1) + try: + ret, num_handles = multi.perform() + if ret != pycurl.E_CALL_MULTI_PERFORM: + if num_handles: + multi.select(0.1) + except pycurl.error as e: + logger.error(f"pycurl multi.perform error: {e}. Continuing with other handles.") + # It's difficult to identify which specific handle failed here. + # Detailed error handling for each handle is done below. - for curl in handles: - data = { - 'status': curl.getinfo(pycurl.RESPONSE_CODE), - 'body': curl.body.getvalue(), - } - result.append(data) - multi.remove_handle(curl) - curl.close() + for i, curl in enumerate(handles): + current_option = option_list[i] + try: + status_code = curl.getinfo(pycurl.RESPONSE_CODE) + body_content = curl.body.getvalue() + data = { + 'status': status_code, + 'body': body_content, + } + result.append(data) + + if status_code != 200: + logger.warning( + f"API call to {current_option[3]}:{current_option[4]}{current_option[1]} failed with status {status_code}. Response: {body_content[:200]}...") + else: + logger.debug( + f"API call to {current_option[3]}:{current_option[4]}{current_option[1]} successful (status {status_code}).") + + except pycurl.error as e: + logger.error( + f"Error getting info/body from pycurl handle for {current_option[3]}:{current_option[4]}{current_option[1]}: {e}") + result.append({'status': -1, 'body': ''}) + finally: + multi.remove_handle(curl) + curl.close() + multi.close() + logger.info(f"Completed {len(option_list)} concurrent REST API calls.") return result def insert_certificate_event(time_range, agent_host, server_id, product, event_name, event_type): - response = requests.post(READ_QUERY_URL, data={ - 'q': "SELECT sum(quantity) FROM event_detection WHERE time >= now()-%s AND agent_host='%s' AND server_id='%s' AND product='%s' AND event_name='%s'" % - (time_range, agent_host, server_id, product, event_name) - }) - if response.status_code == 200: - if 'series' not in response.json()["results"][0]: - requests.post(WRITE_QUERY_URL, - data="event_detection,agent_host=%s,server_id=%s,product=%s,event_name=%s,event_type=%s quantity=1i" % - (agent_host, server_id, product, event_name, event_type)) + """ + Inserts or updates certificate event data in InfluxDB 2.x using Flux for checking existence. + """ + logger.debug(f"Checking InfluxDB for existing event: {event_name} on {agent_host} ({server_id})") + # Flux query to check for existing events + # We query the 'event_detection' bucket for events within the specified time_range + # and filter by tags for uniqueness. + flux_query = f''' + from(bucket: "{INFLUXDB_BUCKET_EVENTS}") + |> range(start: -{time_range}) + |> filter(fn: (r) => r._measurement == "event_detection" and + r.agent_host == "{agent_host}" and + r.server_id == "{server_id}" and + r.product == "{product}" and + r.event_name == "{event_name}") + |> count() + |> yield(name: "result") + ''' -conn = psycopg2.connect(database="cm", - user="postgres", - password="", - host="127.0.0.1", - port="5432") -cur = conn.cursor() + try: + tables = influx_query_api.query(flux_query, org=INFLUXDB_ORG) + event_exists = False + for table in tables: + for record in table.records: + if record.get_value() > 0: # If count is > 0, event exists + event_exists = True + break + if event_exists: + break -cur.execute('''CREATE TABLE IF NOT EXISTS adc_vs_ssl_info( - device_name varchar(32) NOT NULL, - vs_name varchar(512) NOT NULL, - vhost_name varchar(512) NOT NULL, - cert_type varchar(16) NOT NULL, - expiration int, - PRIMARY KEY (device_name, vs_name, cert_type)); - ALTER TABLE adc_vs_ssl_info DROP CONSTRAINT adc_vs_ssl_info_pkey; - ALTER TABLE adc_vs_ssl_info DROP CONSTRAINT adc_vs_ssl_info_device_id_fkey; - ALTER TABLE adc_vs_ssl_info ADD COLUMN IF NOT EXISTS id BIGSERIAL; - ALTER TABLE adc_vs_ssl_info ADD CONSTRAINT adc_vs_ssl_info_pkey PRIMARY KEY (id); - ALTER TABLE adc_vs_ssl_info ADD COLUMN IF NOT EXISTS device_id varchar(64); - ALTER TABLE adc_vs_ssl_info ADD CONSTRAINT adc_vs_ssl_info_device_id_fkey FOREIGN KEY (device_id) - REFERENCES device (id) MATCH SIMPLE - ON UPDATE NO ACTION - ON DELETE NO ACTION; - ''') -cur.execute('delete from adc_vs_ssl_info') -conn.commit() + if not event_exists: + # Create a Point for writing data + point = Point("event_detection") \ + .tag("agent_host", agent_host) \ + .tag("server_id", server_id) \ + .tag("product", product) \ + .tag("event_name", event_name) \ + .tag("event_type", event_type) \ + .field("quantity", 1) # InfluxDB 2.x often prefers integer fields explicitly (1i) for quantities -# ToDo: support for filtered device list instead of all the devices. -cur.execute("select name, ip_address, restapi_port, restapi_username, restapi_password, connection, type, " - "id from device") + influx_write_api.write(bucket=INFLUXDB_BUCKET_EVENTS, org=INFLUXDB_ORG, record=point) + logger.info(f"Inserting new event into InfluxDB: {event_name} for {agent_host}.") + logger.debug(f"Event successfully written to InfluxDB for {event_name} on {agent_host}.") + else: + logger.debug( + f"Event {event_name} for {agent_host} already exists in InfluxDB within {time_range}. Skipping write.") -rows = cur.fetchall() -restapi_options = [] -result = [] -ssl_cli_data = json.dumps({'cmd': 'show ssl host'}) -for row in rows: - if row[6].lower() in ADC_TYPE_LIST and row[5] == 'connected': - url = modify_url("/rest/device_type/cli_extend", row[6]) - restapi_options.append(['POST', url, ssl_cli_data, row[1], row[2], row[3], row[4], 'https', '', row[0], row[7]]) -restapi_result = call_restapi_multi(restapi_options) -cert_restapi_options = [] -for index, rst in enumerate(restapi_result): - if rst['status'] == 200: - rsp = rst['body'] - try: - con = json.loads(rsp) - data = con["contents"] - vs_str_list = data.split("\n") - for vs_str in vs_str_list: - tmp_list = vs_str.split('"') - if len(tmp_list) == 5: - if tmp_list[0] == 'ssl host virtual ': - vs_name = tmp_list[3] - vhost_name = tmp_list[1] - cert_cli_data = json.dumps({'cmd': 'show ssl certificate "%s" "simple"' % (vhost_name)}) - cert_restapi_options.append(['POST', restapi_options[index][1], cert_cli_data, - restapi_options[index][3], restapi_options[index][4], - restapi_options[index][5], restapi_options[index][6], 'https', '', - restapi_options[index][9], vs_name, vhost_name, - restapi_options[index][10]]) - except: - pass -cert_restapi_result = call_restapi_multi(cert_restapi_options) -for index, cert_rst in enumerate(cert_restapi_result): - if cert_rst['status'] == 200: - cert_rsp = cert_rst['body'] - try: - cert_con = json.loads(cert_rsp) - cert_data = cert_con["contents"] - cert_str_list = cert_data.split("\n") - cert_type = "" - for cert_str in cert_str_list: - if cert_str == 'RSA Certificate:': - cert_type = "RSA" - continue - if cert_str == 'ECC Certificate:': - cert_type = "ECC" - continue - if cert_str == 'SM2 Certificate:': - cert_type = "SM2" - continue - - if cert_str[:len('Status')] == "Status": - if cert_type: - if cert_str.find("Expired") != -1: - insert_certificate_event('1d', cert_restapi_options[index][3], - cert_restapi_options[index][10], 'virtualService', - 'CertKeyExpired', 'Exception') - - expiration = -1 - cur.execute("insert into adc_vs_ssl_info (device_id, device_name, vs_name, vhost_name, " - "cert_type, expiration) values ('%s', '%s','%s','%s', '%s', %d) on " - "conflict (id) do update set vhost_name=excluded.vhost_name, " - "expiration=excluded.expiration" % (cert_restapi_options[index][12], - cert_restapi_options[index][9], - cert_restapi_options[index][10], - cert_restapi_options[index][11], - cert_type, - expiration - )) - result.append([cert_restapi_options[index][9], cert_restapi_options[index][10], - cert_restapi_options[index][11], cert_type, "Expired"]) - cert_type = "" - continue + except Exception as e: + logger.error(f"Error communicating with InfluxDB 2.x for event insertion: {e}", exc_info=True) - if cert_str[:len('Days before expiration:')] == "Days before expiration:": - if cert_type: - expiration = int(cert_str[len('Days before expiration:'):]) - result.append([cert_restapi_options[index][9], cert_restapi_options[index][10], - cert_restapi_options[index][11], cert_type, expiration]) - if expiration <= 60: - if expiration <= 1: - insert_certificate_event('1d', cert_restapi_options[index][3], - cert_restapi_options[index][10], 'virtualService', - 'CertKeyExpired_1', 'StatusNotification') - elif expiration <= 7: - insert_certificate_event('7d', cert_restapi_options[index][3], - cert_restapi_options[index][10], 'virtualService', - 'CertKeyExpired_7', 'StatusNotification') - elif expiration <= 30: - insert_certificate_event('30d', cert_restapi_options[index][3], - cert_restapi_options[index][10], 'virtualService', - 'CertKeyExpired_30', 'StatusNotification') - else: - insert_certificate_event('60d', cert_restapi_options[index][3], - cert_restapi_options[index][10], 'virtualService', - 'CertKeyExpired_60', 'StatusNotification') - - cur.execute("insert into adc_vs_ssl_info (device_id, device_name, vs_name, vhost_name, " - "cert_type, expiration) values ('%s', '%s','%s','%s', '%s', %d) on conflict (id) " - "do update set vhost_name=excluded.vhost_name, expiration=excluded.expiration" % ( - cert_restapi_options[index][12], cert_restapi_options[index][9], - cert_restapi_options[index][10], - cert_restapi_options[index][11], - cert_type, expiration)) - cert_type = "" - continue - except: - # print "Cert Response error", cert_rsp - pass - -conn.commit() -conn.close() - - def get_expire(data): + """Helper function for sorting certificate expiration data.""" if data[4] == 'Expired': return -1 return data[4] -result.sort(key=get_expire, reverse=False) -print json.dumps(result) +def main(): + logger.info("Script started: check_devices_ssl_certificate.py") + conn = None + cur = None + try: + # PostgreSQL Connection + pg_password = "Array@123$" + if not pg_password: + logger.critical("PG_PASSWORD environment variable not set. Cannot connect to PostgreSQL.") + sys.exit(1) + conn = psycopg2.connect(database="cm", + user="amp_admin", + password=pg_password, + host="127.0.0.1", + port="5432") + cur = conn.cursor() + logger.info("Connected to PostgreSQL database 'cm'.") + + logger.info("Deleting all existing data from adc_vs_ssl_info table.") + cur.execute('delete from adc_vs_ssl_info') # This line truncates the table every run + conn.commit() + logger.info("adc_vs_ssl_info table truncated.") + + # Fetch Devices + logger.info("Fetching device information from 'device' table.") + cur.execute( + "select name, ip_address, restapi_port, restapi_username, restapi_password, connection, type, id from device") + rows = cur.fetchall() + logger.info(f"Fetched {len(rows)} devices.") + + restapi_options = [] + result = [] + ssl_cli_data = json.dumps({'cmd': 'show ssl host'}) + + # Prepare the first set of API calls (show ssl host) + for row in rows: + device_name, ip_address, restapi_port, restapi_username, restapi_password, connection, device_type, device_id = row + + if device_type.lower() in ADC_TYPE_LIST and connection == 'connected': + url = modify_url("/rest/device_type/cli_extend", device_type) + if url: + restapi_options.append( + ['POST', url, ssl_cli_data, ip_address, restapi_port, restapi_username, restapi_password, + 'https', '', device_name, device_id]) + else: + logger.warning( + f"Skipping device '{device_name}' (ID: {device_id}) due to URL modification failure for type '{device_type}'.") + else: + logger.debug( + f"Skipping device '{device_name}' (ID: {device_id}) - type '{device_type}' not in ADC_TYPE_LIST or connection not 'connected'.") + + if not restapi_options: + logger.info("No connected ADC devices found to query 'show ssl host'. Exiting early.") + return + + # Execute the first set of API calls + restapi_result = call_restapi_multi(restapi_options) + + cert_restapi_options = [] + # Process 'show ssl host' results and prepare second set of API calls (show ssl certificate) + logger.info("Processing 'show ssl host' results and preparing 'show ssl certificate' calls.") + for index, rst in enumerate(restapi_result): + original_options = restapi_options[index] + device_name = original_options[9] + device_id = original_options[10] + + if rst['status'] == 200: + rsp = rst['body'] + try: + con = json.loads(rsp) + data = con.get("contents", "") + if not data: + logger.warning( + f"No 'contents' in 'show ssl host' response for device {device_name} (ID: {device_id}). Response: {rsp[:200]}...") + continue + + vs_str_list = data.split("\n") + for vs_str in vs_str_list: + tmp_list = vs_str.split('"') + if len(tmp_list) == 5 and tmp_list[0] == 'ssl host virtual ': + vs_name = tmp_list[3] + vhost_name = tmp_list[1] + cert_cli_data = json.dumps({'cmd': f'show ssl certificate "{vhost_name}" "simple"'}) + cert_restapi_options.append(['POST', original_options[1], cert_cli_data, + original_options[3], original_options[4], + original_options[5], original_options[6], 'https', '', + device_name, vs_name, vhost_name, device_id]) + elif vs_str.strip(): + logger.debug(f"Unexpected 'show ssl host' line format for {device_name}: {vs_str.strip()}") + except json.JSONDecodeError as e: + logger.error( + f"Failed to parse JSON for 'show ssl host' response from {device_name} (ID: {device_id}): {e}. Raw response: {rsp[:200]}...") + except Exception as e: + logger.exception( + f"An unexpected error occurred processing 'show ssl host' response for {device_name} (ID: {device_id}): {e}") + else: + logger.warning( + f"Skipping 'show ssl host' response processing for {device_name} (ID: {device_id}) due to non-200 status: {rst['status']}.") + + if not cert_restapi_options: + logger.info("No virtual hosts found requiring 'show ssl certificate' calls. Exiting early.") + return + + # Execute the second set of API calls + cert_restapi_result = call_restapi_multi(cert_restapi_options) + + # Process certificate details and insert into DB/InfluxDB + logger.info("Processing 'show ssl certificate' results and inserting into database.") + for index, cert_rst in enumerate(cert_restapi_result): + original_cert_options = cert_restapi_options[index] + device_name = original_cert_options[9] + vs_name = original_cert_options[10] + vhost_name = original_cert_options[11] + device_id = original_cert_options[12] + + if cert_rst['status'] == 200: + cert_rsp = cert_rst['body'] + try: + cert_con = json.loads(cert_rsp) + cert_data = cert_con.get("contents", "") + if not cert_data: + logger.warning( + f"No 'contents' in 'show ssl certificate' response for {vhost_name} on {device_name}. Response: {cert_rsp[:200]}...") + continue + + cert_str_list = cert_data.split("\n") + cert_type = "" + expiration_days = None # Renamed for clarity: expiration will be in days + cert_serial = "unknown" + cert_issuer = "unknown" + cert_subject = "unknown" + cert_not_before = "" + cert_not_after = "" + cert_status = "unknown" # New field for InfluxDB + + for cert_str in cert_str_list: + cert_str = cert_str.strip() + + if cert_str == 'RSA Certificate:': + cert_type = "RSA" + elif cert_str == 'ECC Certificate:': + cert_type = "ECC" + elif cert_str == 'SM2 Certificate:': + cert_type = "SM2" + elif cert_str.startswith('Serial Number:'): + cert_serial = cert_str.split(':', 1)[1].strip() + elif cert_str.startswith('Issuer:'): + cert_issuer = cert_str.split(':', 1)[1].strip() + elif cert_str.startswith('Subject:'): + cert_subject = cert_str.split(':', 1)[1].strip() + elif cert_str.startswith('Not Before:'): + cert_not_before = cert_str.split(':', 1)[1].strip() + elif cert_str.startswith('Not After:'): + cert_not_after = cert_str.split(':', 1)[1].strip() + + elif cert_str.startswith('Status'): + if "Expired" in cert_str: + expiration_days = -1 + cert_status = "Expired" + insert_certificate_event( + datetime.datetime.utcnow().isoformat("T") + "Z", # Current UTC time for event + device_id, # Using device_id as tenant_id for event + 'CertKeyExpired', + cert_serial, cert_issuer, cert_subject, cert_not_before, cert_not_after, + cert_status, + {'agent_host': original_cert_options[3], 'product': 'virtualService', + 'event_type': 'Exception', 'vhost_name': vhost_name} + ) + logger.warning( + f"Certificate for {vhost_name} on {device_name} ({vs_name}) is EXPIRED!") + result.append([device_name, vs_name, vhost_name, cert_type, "Expired"]) + cert_type = "" # Reset for next cert in block + elif "Valid" in cert_str: + cert_status = "Valid" + continue + + elif cert_str.startswith('Days before expiration:'): + if cert_type and expiration_days is None: # Only process if cert_type known and not already expired + try: + expiration_days = int(cert_str[len('Days before expiration:'):].strip()) + logger.info( + f"Certificate for {vhost_name} on {device_name} ({vs_name}) expires in {expiration_days} days.") + result.append([device_name, vs_name, vhost_name, cert_type, expiration_days]) + + event_type_suffix = None + if expiration_days <= 1: + event_type_suffix = '1' + elif expiration_days <= 7: + event_type_suffix = '7' + elif expiration_days <= 30: + event_type_suffix = '30' + elif expiration_days <= 60: + event_type_suffix = '60' + + if event_type_suffix: + insert_certificate_event( + datetime.datetime.utcnow().isoformat("T") + "Z", + # Current UTC time for event + device_id, # Using device_id as tenant_id for event + f'CertKeyExpired_{event_type_suffix}', + cert_serial, cert_issuer, cert_subject, cert_not_before, cert_not_after, + cert_status, + {'agent_host': original_cert_options[3], 'product': 'virtualService', + 'event_type': 'StatusNotification', 'vhost_name': vhost_name} + ) + + except ValueError as ve: + logger.error( + f"Could not parse expiration days from '{cert_str}' for {vhost_name} on {device_name}: {ve}") + expiration_days = None + cert_type = "" # Reset for next cert in block + continue + + # After parsing all lines for a certificate, if we have data, insert/update DB + if expiration_days is not None: + try: + # Using ON CONFLICT (device_id, device_name, vs_name, vhost_name, cert_type) + # for a proper UPSERT logic, as 'id' is a serial. + # Ensure the combination of these fields is UNIQUE in your table schema. + cur.execute( + """ + INSERT INTO adc_vs_ssl_info + (device_id, device_name, vs_name, vhost_name, cert_type, expiration) + VALUES (%s, %s, %s, %s, %s, %s) + ON CONFLICT (device_id, device_name, vs_name, vhost_name, cert_type) DO UPDATE SET + expiration = EXCLUDED.expiration; + """, + (device_id, device_name, vs_name, vhost_name, cert_type or "Unknown", expiration_days) + ) + conn.commit() + logger.debug( + f"Inserted/updated DB for cert {vhost_name} on {device_name} with expiration {expiration_days}.") + except psycopg2.Error as e: + logger.error(f"Failed to insert/update DB for cert {vhost_name} on {device_name}: {e}") + conn.rollback() + elif expiration_days is None: + logger.warning( + f"Could not determine expiration status for certificate on {vhost_name} ({vs_name}) on device {device_name}. Cert data: {cert_data[:200]}...") + + + except json.JSONDecodeError as e: + logger.error( + f"Failed to parse JSON for 'show ssl certificate' response from {vhost_name} on {device_name}: {e}. Raw response: {cert_rsp[:200]}...") + except Exception as e: + logger.exception( + f"An unexpected error occurred processing 'show ssl certificate' response for {vhost_name} on {device_name}: {e}") + else: + logger.warning( + f"Skipping 'show ssl certificate' response processing for {vhost_name} on {device_name} due to non-200 status: {cert_rst['status']}.") + + except psycopg2.Error as e: + logger.critical(f"PostgreSQL connection or query error: {e}") + if conn: + conn.close() + sys.exit(1) + except Exception as e: + logger.critical(f"An unhandled critical error occurred during script execution: {e}", exc_info=True) + sys.exit(1) + finally: + if cur: + cur.close() + if conn: + conn.close() + logger.info("PostgreSQL connection closed.") + if influx_client: + influx_client.close() # Close InfluxDB client connection + logger.info("InfluxDB client connection closed.") + + # Sort and print the final result + result.sort(key=get_expire, reverse=False) + logger.info("Certificate expiration data collection complete. Printing sorted results.") + print(json.dumps(result, indent=2)) + logger.info("Script finished successfully.") + + +if __name__ == "__main__": + main() Index: /branches/amp_4_0/scripts/check_build.py =================================================================== --- /branches/amp_4_0/scripts/check_build.py (revision 2630) +++ /branches/amp_4_0/scripts/check_build.py (working copy) @@ -1,60 +1,184 @@ -#!/usr/bin/python import psycopg2 import os import json import datetime -conn = psycopg2.connect(database="cm", - user="postgres", - password="", - host="127.0.0.1", - port="5432") -cur = conn.cursor() +import logging # Import the logging module +import sys # Import sys for graceful exit if logging fails -cur.execute('''CREATE TABLE IF NOT EXISTS device_build_up_info - (device_name varchar(32) PRIMARY KEY NOT NULL, - current_version varchar(64) NOT NULL, - new_version varchar(64));''') -cur.execute('delete from device_build_up_info') -conn.commit() +# --- Logging Setup --- +LOG_FILE = '/var/log/check_devices_build.log' +LOG_LEVEL = logging.INFO -cur.execute("SELECT app_name, build_version FROM update_list") -rows = cur.fetchall() -build_info = [] -for row in rows: - v_list = row[1].split('.') +# Ensure the log directory exists and is writable +log_dir = os.path.dirname(LOG_FILE) +if not os.path.exists(log_dir): try: - data = { - "num":len(v_list), - "last":int(v_list[-1]), - "pre":".".join(v_list[0:-1]), - "version":row[1] - } - build_info.append(data) - except: - pass + os.makedirs(log_dir, exist_ok=True) # exist_ok=True prevents error if dir already exists + except OSError as e: + # Fallback to basic logging to stderr if directory creation fails + logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') + logging.error(f"Failed to create log directory '{log_dir}': {e}. Exiting.") + sys.exit(1) # Exit immediately if logging setup is critical -cur.execute("SELECT version, name FROM device") -rows = cur.fetchall() +try: + logging.basicConfig( + filename=LOG_FILE, + level=LOG_LEVEL, + format='%(asctime)s - %(levelname)s - %(message)s' + ) +except Exception as e: + # Fallback to basic logging to stderr if file logging setup fails + logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') + logging.error(f"Failed to set up file logging to '{LOG_FILE}': {e}. Logging to stderr.") + sys.exit(1) # Exit if logging setup is critical + +logger = logging.getLogger(__name__) # Get a logger for this module + +# --- Database Connection --- +try: + conn = psycopg2.connect( + database="cm", + user="amp_admin", + password="Array@123$", + host="127.0.0.1", + port="5432" + ) + cur = conn.cursor() + logger.info("Successfully connected to the database.") +except psycopg2.Error as e: + logger.exception("Error connecting to the database:") # Use exception for traceback + sys.exit(1) # Exit the script if the database connection fails + +try: + # Clear existing data before inserting new data + cur.execute('DELETE FROM device_build_up_info;') + conn.commit() + logger.info("Table 'device_build_up_info' ensured and cleared.") +except psycopg2.Error as e: + logger.exception("Error managing 'device_build_up_info' table:") # Use exception for traceback + conn.rollback() # Rollback in case of an error during DDL/DML + conn.close() + sys.exit(1) + +# --- Fetch and Process 'update_list' --- +build_info = [] +try: + cur.execute("SELECT app_name, build_version FROM update_list;") + rows = cur.fetchall() + for row in rows: + version_string = row[1] + if version_string is None: # Handle cases where build_version might be NULL + logger.warning(f"Skipping update_list entry for app '{row[0]}' due to NULL build_version.") + continue + + v_list = version_string.split('.') + try: + # Ensure the last part can be converted to an integer + last_part_int = int(v_list[-1]) + data = { + "num": len(v_list), + "last": last_part_int, + "pre": ".".join(v_list[0:-1]), + "version": version_string + } + build_info.append(data) + logger.debug(f"Processed update_list entry: {data}") + except ValueError: + logger.warning(f"Could not parse version part for '{version_string}'. Skipping.") + pass # Pass if the last part is not an integer (e.g., "1.2.a") +except psycopg2.Error as e: + logger.exception("Error fetching data from 'update_list':") # Use exception for traceback + conn.close() + sys.exit(1) + +# --- Fetch and Process 'device' data, then Insert into 'device_build_up_info' --- result = [] -for row in rows: - try: - verion_str = row[0].strip().split('\n')[0] - version = verion_str.split(' ')[1] - version_list = version.split('.') - com_v = ".".join(version_list[:-1]) - new_ver = "" - curr_ver = int(version_list[-1]) - for k in build_info: - if k["pre"] == com_v and k["num"] == len(version_list): - if k["last"] > curr_ver: - curr_ver = k["last"] - new_ver = k["version"] - cur.execute("insert into device_build_up_info values ('%s','%s','%s')" - "on conflict (device_name) do update set current_version=excluded.current_version," - "new_version=excluded.new_version" %(row[1],version,new_ver)) - result.append([row[1],version,new_ver]) - except: - pass -conn.commit() -conn.close() -print json.dumps(result) +try: + cur.execute("SELECT version, name FROM device;") + rows = cur.fetchall() + for row in rows: + device_name = row[1] + version_raw = row[0] # Get raw version string from DB + + if version_raw is None: # Handle cases where device version might be NULL + logger.warning(f"Skipping device '{device_name}' due to NULL version.") + continue + + version_raw = version_raw.strip() # Strip whitespace from raw version + + actual_version_string = "" + # If the version string might contain spaces (e.g., "v 1.2.3"), split by space + # Otherwise, just use version_raw directly if it's always "1.2.3" + if ' ' in version_raw: + version_parts = version_raw.split(' ') + # Assuming the actual version is the second part + if len(version_parts) > 1: + actual_version_string = version_parts[1] + else: + logger.warning(f"Unexpected version format '{version_raw}' for device '{device_name}'. Skipping.") + continue + else: + actual_version_string = version_raw + + try: + version_list = actual_version_string.split('.') + if not version_list or not version_list[-1].isdigit(): + raise ValueError("Last part of version is not a digit or version is empty.") + + common_version_prefix = ".".join(version_list[:-1]) + current_device_last_part = int(version_list[-1]) # Last integer part of current device version + + new_highest_version = None # Initialize to None + # Find the highest compatible version from build_info + for k in build_info: + if k["pre"] == common_version_prefix and k["num"] == len(version_list): + if k["last"] > current_device_last_part: + current_device_last_part = k["last"] # Update to the higher last part found + new_highest_version = k["version"] # Store the full new version string + + # If no newer version is found, new_highest_version remains None. + # In your schema, new_version can be NULL. + + # Use parameterized queries to prevent SQL injection vulnerabilities + # and handle quoting automatically. + cur.execute( + """ + INSERT INTO device_build_up_info (device_name, current_version, new_version) + VALUES (%s, %s, %s) + ON CONFLICT (device_name) DO UPDATE SET + current_version = EXCLUDED.current_version, + new_version = EXCLUDED.new_version; + """, + (device_name, actual_version_string, new_highest_version) + ) + result.append([device_name, actual_version_string, new_highest_version]) + logger.debug( + f"Processed device '{device_name}': current='{actual_version_string}', new='{new_highest_version}'") + + except (ValueError, IndexError) as e: + logger.warning( + f"Error parsing version for device '{device_name}' with raw version '{version_raw}': {e}. Skipping this device.") + pass # Skip this row if parsing fails + except psycopg2.Error as e: + logger.exception(f"Error inserting/updating device_build_up_info for device '{device_name}':") + # Do not roll back for individual row errors if you want to continue processing others + pass + + conn.commit() # Commit all insertions/updates at once + logger.info("Device build up information processed and inserted/updated.") + +except psycopg2.Error as e: + logger.exception("Error fetching data from 'device' table:") + conn.rollback() # Rollback if fetching 'device' data itself fails + sys.exit(1) +finally: + # Always ensure the connection is closed + if conn: + cur.close() + conn.close() + logger.info("Database connection closed.") + +# --- Output Results --- +print(json.dumps(result, indent=4)) + +logger.info("Script finished successfully.") Index: /branches/amp_4_0/scripts/check_crontab_add.py =================================================================== --- /branches/amp_4_0/scripts/check_crontab_add.py (revision 2630) +++ /branches/amp_4_0/scripts/check_crontab_add.py (working copy) @@ -1,59 +1,137 @@ import subprocess import os +import logging +# --- Logging Setup --- +LOG_FILE = '/var/log/check_devices_ssl_certificate.log' +LOG_LEVEL = logging.INFO # Set to logging.DEBUG for more verbose output -def _runcmd(cmd, input=None): +# Ensure the log directory exists and is writable (best practice for production) +log_dir = os.path.dirname(LOG_FILE) +if not os.path.exists(log_dir): + try: + os.makedirs(log_dir, exist_ok=True) # exist_ok=True prevents error if dir already exists + except OSError as e: + # Fallback to stderr if we can't create the log directory + logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') + logging.error(f"Failed to create log directory {log_dir}: {e}. Logging to stderr.") + # Re-raise to stop if logging setup is critical + raise - if input is not None: - p = subprocess.Popen(cmd, - shell=True, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - close_fds=True, - preexec_fn=os.setsid) - else: - p = subprocess.Popen(cmd, - shell=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - close_fds=True, - preexec_fn=os.setsid) +try: + logging.basicConfig( + filename=LOG_FILE, + level=LOG_LEVEL, + format='%(asctime)s - %(levelname)s - %(message)s' + ) +except Exception as e: + # Fallback to stderr if file logging fails (e.g., permissions) + logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s') + logging.error(f"Failed to set up file logging to {LOG_FILE}: {e}. Logging to stderr.") - stdoutdata, stderrdata = p.communicate(input) - return p.returncode, stderrdata, stdoutdata +logger = logging.getLogger(__name__) -def deploycrontab(): +def _runcmd(cmd, input_data=None): + """ + Executes a shell command and captures its output and error. + """ + try: + process = subprocess.Popen( + cmd, + shell=True, + stdin=subprocess.PIPE if input_data is not None else None, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + close_fds=True, + preexec_fn=os.setsid, # Detach from controlling terminal + text=True # Decode stdout/stderr as text + ) + + stdout_data, stderr_data = process.communicate(input_data) + return process.returncode, stderr_data, stdout_data + except Exception as e: + logger.error(f"Error executing command '{cmd}': {e}") + raise + + +def deploy_crontab(): + """ + Deploys and manages specific cron jobs by reading existing ones, + filtering out certain entries, and adding new ones. + """ + logger.info("Starting crontab deployment.") + + # 1. Read existing crontab retcode, err, installed_content = _runcmd("crontab -l") - if retcode != 0 and 'no crontab for' not in err: - raise OSError("crontab not supported in your system") - installed_content = installed_content.rstrip("\n") - installed_crontabs = installed_content.split("\n") - installed_content = "" - for installed_crontab in installed_crontabs: - if installed_crontab: - if "/ca/bin/check_" in installed_crontab: - pass - else: - if not installed_content: - installed_content += installed_crontab - else: - installed_content += "\n%s" % installed_crontab - content = "30 0 * * * /usr/bin/python /ca/bin/check_build.py\n0 1 * * * /usr/bin/python /ca/bin/check_adc_ssl.py\n*/30 * * * * sh /ca/bin/check_pg_conn.sh\n*/10 * * * * sh /ca/bin/check_rsync.sh" - if not installed_content: - installed_content += content + if retcode != 0: + if 'no crontab for' in err: + logger.info("No existing crontab found for the current user. Starting fresh.") + installed_content = "" # Initialize as empty string + else: + logger.error(f"Failed to list crontab. Return code: {retcode}, Error: {err.strip()}") + raise OSError("Crontab not supported or accessible on your system.") else: - installed_content += "\n%s" % content + logger.info("Existing crontab content retrieved.") + logger.debug(f"Raw existing crontab:\n{installed_content.strip()}") - if installed_content: - installed_content += "\n" - # install back - retcode, err, out = _runcmd("crontab", installed_content) + # 2. Filter out specific cron entries + # Split content into lines, remove empty lines, and filter based on the keyword + installed_lines = installed_content.strip().split("\n") if installed_content.strip() else [] + + kept_crontabs = [] + filtered_out_count = 0 + for line in installed_lines: + if line and "/ca/bin/check_" in line: + logger.info(f"Filtering out existing cron job: {line.strip()}") + filtered_out_count += 1 + else: + kept_crontabs.append(line) + + current_crontab_to_keep = "\n".join(kept_crontabs) + + if filtered_out_count > 0: + logger.info(f"Filtered out {filtered_out_count} existing cron jobs.") + else: + logger.info("No existing '/ca/bin/check_' cron jobs found to filter.") + + # 3. Define the new cron jobs to be added + # This could be read from a config file for more flexibility + # ToDo: Find RCA & Fix rsync, it's not working in AMP3 as well - "*/10 * * * * sh /ca/bin/check_rsync.sh" + # ToDo: Test & Enable - "0 1 * * * /usr/bin/python3 /ca/bin/check_adc_ssl.py", + new_cron_jobs = [ + "30 0 * * * /usr/bin/python3 /ca/bin/check_build.py", + ] + content_to_add = "\n".join(new_cron_jobs) + logger.info("New cron jobs to be added:\n" + content_to_add) + + # 4. Combine the filtered existing jobs with the new jobs + final_crontab_content = current_crontab_to_keep + if final_crontab_content: # If there's existing content to keep + final_crontab_content += "\n" # Add a newline separator + final_crontab_content += content_to_add + + # Ensure a trailing newline is present, which crontab expects + if not final_crontab_content.endswith("\n"): + final_crontab_content += "\n" + + logger.debug(f"Final crontab content to install:\n{final_crontab_content.strip()}") + + # 5. Install the new crontab + retcode, err, out = _runcmd("crontab", input_data=final_crontab_content) + if retcode != 0: - raise ValueError( - "failed to install crontab, check if crontab is valid") + logger.error(f"Failed to install crontab. Return code: {retcode}, Error: {err.strip()}") + raise ValueError("Failed to install crontab. Check if crontab syntax is valid or permissions.") + else: + logger.info("Crontab deployed successfully.") + logger.debug(f"Crontab command output: {out.strip()}") -deploycrontab() +# Execute the deployment +if __name__ == "__main__": + try: + deploy_crontab() + except Exception as e: + logger.critical(f"Script failed due to an unhandled error: {e}")