# http_client.py # # Copyright 2025 Pavel Baksy # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # # SPDX-License-Identifier: GPL-3.0-or-later import gi gi.require_version('Soup', '3.0') from gi.repository import Soup, GLib, Gio import time from typing import Optional, Callable, Any from .models import HttpRequest, HttpResponse class HttpClient: """HTTP client using libsoup3 async API.""" def __init__(self): """Initialize HTTP client with reusable session.""" self.session = Soup.Session.new() # Load settings self.settings = Gio.Settings.new('cz.vesp.roster') # Initialize TLS verification flag self.force_tls_verification = True # Apply initial settings self._apply_settings() # Listen for settings changes self.settings.connect('changed::force-tls-verification', self._on_settings_changed) self.settings.connect('changed::request-timeout', self._on_settings_changed) def _apply_settings(self): """Apply settings to the HTTP session.""" # Store TLS verification preference (will be applied per-message) self.force_tls_verification = self.settings.get_boolean('force-tls-verification') # Apply timeout setting timeout = self.settings.get_int('request-timeout') self.session.set_timeout(timeout) def _on_settings_changed(self, settings, key): """Handle settings changes.""" self._apply_settings() def _accept_all_certificates(self, msg, tls_certificate, tls_errors): """Accept all TLS certificates when verification is disabled.""" return True # Accept certificate regardless of errors def execute_request_async(self, request: HttpRequest, callback: Callable, user_data: Any = None): """ Execute HTTP request asynchronously using libsoup. Args: request: HttpRequest object with method, url, headers, body callback: Function to call with results: callback(response, error, user_data) - response: HttpResponse object if successful, None on error - error: Error message string if failed, None on success - user_data: The user_data passed to this function user_data: Optional data to pass to callback """ start_time = time.time() # Validate and create message if not request.url or not request.url.strip(): callback(None, "Invalid URL: URL is empty", user_data) return try: msg = Soup.Message.new(request.method, request.url) except Exception as e: callback(None, f"Invalid URL: {e}", user_data) return # Handle TLS certificate verification if not self.force_tls_verification: # Connect signal to accept all certificates when verification is disabled msg.connect('accept-certificate', self._accept_all_certificates) # Add headers headers = msg.get_request_headers() for key, value in request.headers.items(): if key and value: # Skip empty headers headers.append(key, value) # Set body for POST/PUT/DELETE if request.method in ['POST', 'PUT', 'DELETE'] and request.body.strip(): content_type = self._get_content_type(request) msg.set_request_body_from_bytes( content_type, GLib.Bytes.new(request.body.encode('utf-8')) ) # Define callback wrapper to handle timing and parsing def on_response_ready(session, result, user_data_tuple): cb, udata, start = user_data_tuple response_time = (time.time() - start) * 1000 # Convert to ms try: # Finish async operation bytes_data = session.send_and_read_finish(result) # Parse response from message response = self._parse_response(msg, bytes_data, response_time) cb(response, None, udata) except GLib.Error as e: # Handle libsoup errors error_msg = self._format_error(e) cb(None, error_msg, udata) # Send async request self.session.send_and_read_async( msg, GLib.PRIORITY_DEFAULT, None, # cancellable on_response_ready, (callback, user_data, start_time) ) def _parse_response(self, msg: Soup.Message, bytes_data: GLib.Bytes, response_time: float) -> HttpResponse: """ Extract response data from Soup.Message. Args: msg: The Soup.Message with response headers and status bytes_data: Response body as GLib.Bytes response_time: Response time in milliseconds Returns: HttpResponse object with status, headers, body, and timing """ # Extract status code and text status_code = msg.get_status() status_text = msg.get_reason_phrase() # Format headers in HTTPie-style output (HTTP/1.1 200 OK\nHeader: value\n...) headers_text = self._format_headers(msg, status_code, status_text) # Decode body with error handling body = bytes_data.get_data().decode('utf-8', errors='replace') return HttpResponse( status_code=status_code, status_text=status_text, headers=headers_text, body=body.strip(), response_time_ms=response_time ) def _format_headers(self, msg: Soup.Message, status_code: int, status_text: str) -> str: """ Format response headers in HTTPie-style output format. Format: HTTP/1.1 200 OK Header-Name: value Another-Header: value Args: msg: The Soup.Message to extract headers from status_code: HTTP status code status_text: HTTP status text (e.g., "OK") Returns: Formatted header string """ # Start with status line lines = [f"HTTP/1.1 {status_code} {status_text}"] # Get response headers and iterate headers_dict = msg.get_response_headers() # Use foreach to iterate through headers def append_header(name, value, lines_list): lines_list.append(f"{name}: {value}") headers_dict.foreach(append_header, lines) return '\n'.join(lines) def _get_content_type(self, request: HttpRequest) -> str: """ Determine content-type for request body. Args: request: HttpRequest with headers Returns: Content-Type string (defaults to text/plain) """ # Check if user specified Content-Type header for key, value in request.headers.items(): if key.lower() == 'content-type': return value # Default to plain text return "text/plain" def _format_error(self, error: GLib.Error) -> str: """ Convert GLib.Error to user-friendly message. Args: error: GLib.Error from libsoup Returns: User-friendly error message string """ error_str = str(error).lower() if "timeout" in error_str: return "Request timeout (30 seconds)" elif "resolve" in error_str: return f"Could not resolve hostname: {error}" elif "connect" in error_str: return f"Connection failed: {error}" else: return f"Request failed: {error}"