206 lines
9.2 KiB
Python
206 lines
9.2 KiB
Python
"""
|
|
HTTP Utilities node for ComfyUI
|
|
Provides various HTTP utility functions
|
|
"""
|
|
|
|
import json
|
|
import urllib.parse
|
|
from typing import Dict, Any, Tuple
|
|
import re
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import time
|
|
|
|
class HTTPUtils:
|
|
"""HTTP Utilities Node for ComfyUI"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"operation": ([
|
|
"url_encode", "url_decode", "base64_encode", "base64_decode",
|
|
"json_escape", "json_unescape", "html_escape", "html_unescape",
|
|
"generate_timestamp", "generate_uuid", "generate_signature",
|
|
"parse_url", "build_url", "extract_domain", "validate_email",
|
|
"generate_bearer_token", "create_basic_auth"
|
|
], {"default": "url_encode"}),
|
|
"input_data": ("STRING", {"default": ""}),
|
|
},
|
|
"optional": {
|
|
"secret_key": ("STRING", {"default": ""}),
|
|
"algorithm": (["sha1", "sha256", "md5"], {"default": "sha256"}),
|
|
"url_base": ("STRING", {"default": "https://api.example.com"}),
|
|
"url_path": ("STRING", {"default": "/endpoint"}),
|
|
"url_params": ("STRING", {"default": "{}"}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING", "STRING", "BOOLEAN")
|
|
RETURN_NAMES = ("result", "info", "success")
|
|
FUNCTION = "execute_utility"
|
|
CATEGORY = "HTTP/Utilities"
|
|
|
|
def execute_utility(self, operation: str, input_data: str, secret_key: str = "",
|
|
algorithm: str = "sha256", url_base: str = "https://api.example.com",
|
|
url_path: str = "/endpoint", url_params: str = "{}") -> Tuple[str, str, bool]:
|
|
"""Execute various HTTP utility operations"""
|
|
|
|
try:
|
|
if operation == "url_encode":
|
|
result = urllib.parse.quote(input_data, safe='')
|
|
info = f"URL encoded {len(input_data)} characters"
|
|
return (result, info, True)
|
|
|
|
elif operation == "url_decode":
|
|
result = urllib.parse.unquote(input_data)
|
|
info = f"URL decoded to {len(result)} characters"
|
|
return (result, info, True)
|
|
|
|
elif operation == "base64_encode":
|
|
encoded = base64.b64encode(input_data.encode('utf-8')).decode('utf-8')
|
|
info = f"Base64 encoded {len(input_data)} characters to {len(encoded)} characters"
|
|
return (encoded, info, True)
|
|
|
|
elif operation == "base64_decode":
|
|
try:
|
|
decoded = base64.b64decode(input_data).decode('utf-8')
|
|
info = f"Base64 decoded {len(input_data)} characters to {len(decoded)} characters"
|
|
return (decoded, info, True)
|
|
except Exception as e:
|
|
return (f"Decode error: {str(e)}", f"Base64 decode failed", False)
|
|
|
|
elif operation == "json_escape":
|
|
escaped = json.dumps(input_data)
|
|
info = f"JSON escaped string"
|
|
return (escaped, info, True)
|
|
|
|
elif operation == "json_unescape":
|
|
try:
|
|
unescaped = json.loads(input_data)
|
|
info = f"JSON unescaped string"
|
|
return (str(unescaped), info, True)
|
|
except Exception as e:
|
|
return (f"Unescape error: {str(e)}", f"JSON unescape failed", False)
|
|
|
|
elif operation == "html_escape":
|
|
import html
|
|
escaped = html.escape(input_data)
|
|
info = f"HTML escaped {len(input_data)} characters"
|
|
return (escaped, info, True)
|
|
|
|
elif operation == "html_unescape":
|
|
import html
|
|
unescaped = html.unescape(input_data)
|
|
info = f"HTML unescaped {len(input_data)} characters"
|
|
return (unescaped, info, True)
|
|
|
|
elif operation == "generate_timestamp":
|
|
timestamp = str(int(time.time()))
|
|
iso_timestamp = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
|
|
info = f"Generated timestamp: {iso_timestamp}"
|
|
return (timestamp, info, True)
|
|
|
|
elif operation == "generate_uuid":
|
|
import uuid
|
|
generated_uuid = str(uuid.uuid4())
|
|
info = f"Generated UUID v4"
|
|
return (generated_uuid, info, True)
|
|
|
|
elif operation == "generate_signature":
|
|
if not secret_key:
|
|
return ("Error: Secret key required", "Missing secret key", False)
|
|
|
|
if algorithm == "sha1":
|
|
signature = hmac.new(secret_key.encode(), input_data.encode(), hashlib.sha1).hexdigest()
|
|
elif algorithm == "sha256":
|
|
signature = hmac.new(secret_key.encode(), input_data.encode(), hashlib.sha256).hexdigest()
|
|
elif algorithm == "md5":
|
|
signature = hmac.new(secret_key.encode(), input_data.encode(), hashlib.md5).hexdigest()
|
|
else:
|
|
return ("Error: Unsupported algorithm", "Invalid algorithm", False)
|
|
|
|
info = f"Generated {algorithm.upper()} HMAC signature"
|
|
return (signature, info, True)
|
|
|
|
elif operation == "parse_url":
|
|
try:
|
|
parsed = urllib.parse.urlparse(input_data)
|
|
result = json.dumps({
|
|
"scheme": parsed.scheme,
|
|
"netloc": parsed.netloc,
|
|
"path": parsed.path,
|
|
"params": parsed.params,
|
|
"query": parsed.query,
|
|
"fragment": parsed.fragment,
|
|
"hostname": parsed.hostname,
|
|
"port": parsed.port,
|
|
"username": parsed.username,
|
|
"password": parsed.password
|
|
}, indent=2)
|
|
info = f"Parsed URL components"
|
|
return (result, info, True)
|
|
except Exception as e:
|
|
return (f"Parse error: {str(e)}", "URL parsing failed", False)
|
|
|
|
elif operation == "build_url":
|
|
try:
|
|
params_dict = json.loads(url_params) if url_params and url_params != "{}" else {}
|
|
|
|
# Build base URL
|
|
base = url_base.rstrip('/')
|
|
path = url_path.lstrip('/')
|
|
full_url = f"{base}/{path}"
|
|
|
|
# Add query parameters if any
|
|
if params_dict:
|
|
query_string = urllib.parse.urlencode(params_dict)
|
|
full_url += f"?{query_string}"
|
|
|
|
info = f"Built URL with {len(params_dict)} parameters"
|
|
return (full_url, info, True)
|
|
except Exception as e:
|
|
return (f"Build error: {str(e)}", "URL building failed", False)
|
|
|
|
elif operation == "extract_domain":
|
|
try:
|
|
parsed = urllib.parse.urlparse(input_data)
|
|
domain = parsed.netloc or parsed.path.split('/')[0]
|
|
# Remove port if present
|
|
domain = domain.split(':')[0]
|
|
info = f"Extracted domain from URL"
|
|
return (domain, info, True)
|
|
except Exception as e:
|
|
return (f"Extract error: {str(e)}", "Domain extraction failed", False)
|
|
|
|
elif operation == "validate_email":
|
|
email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'
|
|
is_valid = bool(re.match(email_pattern, input_data))
|
|
info = f"Email validation: {'valid' if is_valid else 'invalid'}"
|
|
return (str(is_valid).lower(), info, is_valid)
|
|
|
|
elif operation == "generate_bearer_token":
|
|
# Simple bearer token generation (in practice, use proper JWT or OAuth)
|
|
token_data = f"{input_data}:{int(time.time())}"
|
|
token = base64.b64encode(token_data.encode()).decode()
|
|
bearer_token = f"Bearer {token}"
|
|
info = f"Generated bearer token"
|
|
return (bearer_token, info, True)
|
|
|
|
elif operation == "create_basic_auth":
|
|
# Create basic auth header from username:password
|
|
if ':' not in input_data:
|
|
return ("Error: Input should be 'username:password'", "Invalid format", False)
|
|
|
|
encoded = base64.b64encode(input_data.encode()).decode()
|
|
basic_auth = f"Basic {encoded}"
|
|
info = f"Created basic auth header"
|
|
return (basic_auth, info, True)
|
|
|
|
else:
|
|
return (f"Unknown operation: {operation}", "Invalid operation", False)
|
|
|
|
except Exception as e:
|
|
error_msg = f"Utility operation failed: {str(e)}"
|
|
return (error_msg, error_msg, False) |