import os import sys import json import logging import requests import csv from requests.exceptions import RequestException import urllib3 from time import sleep urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) logging.basicConfig( format='%(asctime)s %(levelname)s %(message)s', datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO ) class KicsApiClient: def __init__(self, metadata_path, cert_path, key_path): self.metadata_path = metadata_path self.cert_path = cert_path self.key_path = key_path self.base_url = None self.token = None self._initialize() def _initialize(self): try: with open(self.metadata_path, 'r', encoding='utf-8') as f: metadata = json.load(f) if 'serverUri' not in metadata: raise ValueError("Invalid metadata format") self.base_url = metadata['serverUri'].rstrip('/') if not self.base_url.endswith('/kics/api'): self.base_url += '/kics/api' logging.info(f"API Base URL: {self.base_url}") self._authenticate() except Exception as e: logging.error(f"Initialization error: {str(e)}") sys.exit(1) def _authenticate(self): try: response = requests.post( f"{self.base_url}/auth/token?grant_type=certificate", cert=(self.cert_path, self.key_path), verify=False ) response.raise_for_status() self.token = response.json().get('access_token') if not self.token: raise ValueError("Authentication failed") logging.info("Authentication successful") except RequestException as e: logging.error(f"Auth error: {str(e)}") sys.exit(1) def get_all_allow_rules(self, batch_size=1000, delay=1): all_rules = [] offset = 0 total_processed = 0 while True: try: batch = self._get_rules_batch(offset, batch_size) if not batch: break all_rules.extend(batch) total_processed += len(batch) logging.info(f"Received {len(batch)} rules, total: {total_processed}") if len(batch) < batch_size: break offset += batch_size sleep(delay) except Exception as e: logging.error(f"Error fetching batch {offset}-{offset+batch_size}: {str(e)}") break return all_rules def _get_rules_batch(self, offset, limit): url = f"{self.base_url}/v4/allow-rules/query" headers = { 'Authorization': f'Bearer {self.token}', 'Content-Type': 'application/json' } query = { "sort": [ { "column": "Id", "direction": "Asc" } ], "offset": offset, "limit": limit } try: response = requests.post( url, headers=headers, json=query, verify=False, timeout=30 ) response.raise_for_status() return response.json().get('values', []) except RequestException as e: self._handle_error(e) return [] def _handle_error(self, error): error_msg = "API Error:" if error.response: error_msg += f"\nStatus: {error.response.status_code}" error_msg += f"\nResponse: {error.response.text[:500]}" logging.error(error_msg) class CsvExporter: @staticmethod def export(rules, filename, batch_size=10000): if not rules: logging.warning("No data to export") return try: with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f, delimiter=';', quoting=csv.QUOTE_ALL) headers = [ 'ID', 'Type', 'Comment', 'Created', 'Modified', 'Monitoring Point', 'Event Type', 'Triggered Rule', 'Is Active', 'Is DPI Detectable', 'Address Type', 'Commands', 'Protocols', 'Side1 MAC Ranges', 'Side1 IP Ranges', 'Side1 Port Ranges', 'Side1 MAC Spaces', 'Side1 IP Spaces', 'Side2 MAC Ranges', 'Side2 IP Ranges', 'Side2 Port Ranges', 'Side2 MAC Spaces', 'Side2 IP Spaces', 'Is Auto Generated' ] writer.writerow(headers) for i in range(0, len(rules), batch_size): batch = rules[i:i+batch_size] rows = [] for rule in batch: try: rows.append(CsvExporter._process_rule(rule)) except Exception as e: rule_id = CsvExporter._safe_get(rule, 'id', str, 'unknown') logging.error(f"Error processing rule {rule_id}: {str(e)}") continue writer.writerows(rows) logging.info(f"Exported {len(rows)} rules to {filename} (total: {i+len(rows)})") logging.info(f"Successfully exported {len(rules)} rules to {filename}") except Exception as e: logging.error(f"Export error: {str(e)}") sys.exit(1) @staticmethod def _process_rule(rule): row = [] row.append(CsvExporter._safe_get(rule, 'id', int)) row.append(CsvExporter._safe_get(rule, 'ruleType', str)) row.append(CsvExporter._safe_get(rule, 'comment', str)) row.append(CsvExporter._safe_get(rule, 'timestampCreated', str)) row.append(CsvExporter._safe_get(rule, 'timestampModified', str)) row.append(CsvExporter._safe_get(rule, 'monitoringPoint', str)) row.append(CsvExporter._safe_get(rule, 'eventType', str)) row.append(CsvExporter._safe_get(rule, 'triggeredRule', str)) row.append(CsvExporter._safe_get(rule, 'isActive', bool, False)) row.append(CsvExporter._safe_get(rule, 'isDpiDetectable', bool, False)) row.append(CsvExporter._safe_get(rule, 'addressType', str)) row.append(CsvExporter._parse_commands(rule)) row.append(CsvExporter._safe_get(rule, 'protocols', str)) side1 = CsvExporter._safe_get(rule, 'side1', dict, {}) row.append(CsvExporter._parse_ranges(side1.get('macAddressRanges'))) row.append(CsvExporter._parse_ranges(side1.get('ipAddressRanges'))) row.append(CsvExporter._parse_ranges(side1.get('portAddressRanges'))) row.append(CsvExporter._parse_ids(side1.get('macAddressSpaceIds'))) row.append(CsvExporter._parse_ids(side1.get('ipAddressSpaceIds'))) side2 = CsvExporter._safe_get(rule, 'side2', dict, {}) row.append(CsvExporter._parse_ranges(side2.get('macAddressRanges'))) row.append(CsvExporter._parse_ranges(side2.get('ipAddressRanges'))) row.append(CsvExporter._parse_ranges(side2.get('portAddressRanges'))) row.append(CsvExporter._parse_ids(side2.get('macAddressSpaceIds'))) row.append(CsvExporter._parse_ids(side2.get('ipAddressSpaceIds'))) row.append(CsvExporter._safe_get(rule, 'isAutoGenerated', bool, False)) return row @staticmethod def _parse_commands(rule): commands = CsvExporter._safe_get(rule, 'commands', str, '') return commands.replace('; ', ';').replace(';', '; ') @staticmethod def _parse_ranges(ranges): try: if not isinstance(ranges, list): return '' return ', '.join( [f"{r.get('from', '')}-{r.get('to', '')}" if r.get('from') != r.get('to') else str(r.get('from', '')) for r in ranges if isinstance(r, dict)] ) except: return '' @staticmethod def _parse_ids(ids): try: if not isinstance(ids, list): return '' return ', '.join(map(str, ids)) except: return '' @staticmethod def _safe_get(data, key, expected_type, default=None): try: value = data.get(key, default) if value is None: return default if expected_type == bool: return bool(value) if isinstance(value, expected_type): return value try: return expected_type(value) except: return default except Exception as e: return default def main(): BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CONNECTOR_NAME = "коннектор_GetAR" PATHS = { 'metadata': os.path.join(BASE_DIR, 'connector', CONNECTOR_NAME, 'metadata.json'), 'cert': '/var/opt/kaspersky/kics4net-connectors/instances/GetAR/ssl/connector.crt', 'key': '/var/opt/kaspersky/kics4net-connectors/instances/GetAR/ssl/connector.key' } for path in PATHS.values(): if not os.path.exists(path): logging.error(f"File missing: {path}") return try: api = KicsApiClient(PATHS['metadata'], PATHS['cert'], PATHS['key']) logging.info("Starting full export...") rules = api.get_all_allow_rules(batch_size=1000) logging.info(f"Total rules received: {len(rules)}") CsvExporter.export(rules, 'allow_rules_full.csv') except Exception as e: logging.error(f"Critical error: {str(e)}") if __name__ == '__main__': main()