3bc155b127
- Add PreferencesDialog with settings for TLS verification and request timeout - Update HttpClient to respect TLS verification setting - Add GSettings schema keys for force-tls-verification and request-timeout - Wire up preferences action to show dialog - Settings changes apply immediately to HTTP session
236 lines
8.0 KiB
Python
236 lines
8.0 KiB
Python
# 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 <https://www.gnu.org/licenses/>.
|
|
#
|
|
# 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 like HTTPie 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 to match HTTPie 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}"
|