refactor: reduce complexity in latitude and ovh cloud libs (#835)

- latitude/lib/common.sh: Replace custom 38-line wait_for_server_ready
  polling loop with generic_wait_for_instance from shared/common.sh.
  Consolidate extract_latitude_server_ip (36 lines of inline Python) into
  a single readonly expression constant. Net -59 lines.

- ovh/lib/common.sh: Replace shell variable interpolation in Python
  strings ('${var}') with sys.argv[] in _ovh_find_flavor_id,
  _ovh_get_ssh_key_id, _ovh_build_instance_body, and ovh_register_ssh_key.
  This eliminates injection surface and follows the established pattern
  used by other cloud providers.

Agent: complexity-hunter

Co-authored-by: A <6723574+louisgv@users.noreply.github.com>
Co-authored-by: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
A 2026-02-13 01:17:20 -08:00 committed by GitHub
parent ac3a8c58a5
commit cb2a8614e9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 26 additions and 85 deletions

View file

@ -259,82 +259,24 @@ create_server() {
log_step "Waiting for server provisioning (this may take a few minutes for bare metal)..."
}
# Extract the IPv4 address from a Latitude.sh server API response
# Checks network.ip, ip_addresses[], and primary_ipv4 fields
# Usage: extract_latitude_server_ip JSON_RESPONSE
extract_latitude_server_ip() {
local response="$1"
echo "$response" | python3 -c "
import json, sys
data = json.loads(sys.stdin.read())
server = data.get('data', {})
attrs = server.get('attributes', {})
# Check for IP in network attributes
network = attrs.get('network', {})
if isinstance(network, dict):
ip = network.get('ip', '')
if ip:
print(ip)
sys.exit(0)
# Check for IP in relationships or included data
ips = attrs.get('ip_addresses', [])
if isinstance(ips, list):
for ip_obj in ips:
if isinstance(ip_obj, dict):
addr = ip_obj.get('address', '')
if addr and ':' not in addr: # Skip IPv6
print(addr)
sys.exit(0)
elif isinstance(ip_obj, str) and ':' not in ip_obj:
print(ip_obj)
sys.exit(0)
# Fallback: try primary_ipv4
primary = attrs.get('primary_ipv4', '')
if primary:
print(primary)
sys.exit(0)
sys.exit(1)
" 2>/dev/null
}
# Python expression to extract IPv4 from Latitude.sh JSON:API response.
# Checks: network.ip, ip_addresses[] (dict or string, skip IPv6), primary_ipv4.
# Used by generic_wait_for_instance; receives 'd' as the parsed JSON dict.
readonly _LATITUDE_IP_PY="(lambda a: (a.get('network',{}).get('ip','') if isinstance(a.get('network'),dict) else '') or next((o.get('address','') if isinstance(o,dict) else o for o in (a.get('ip_addresses') or []) if ':' not in (o.get('address','') if isinstance(o,dict) else o)),None) or a.get('primary_ipv4',''))(d.get('data',{}).get('attributes',{}))"
# Wait for server to become active and get its IP address
# Delegates to generic_wait_for_instance from shared/common.sh.
# Latitude reports status as "on" when active, so we match on "on".
wait_for_server_ready() {
local server_id="$1"
local max_attempts=${2:-60}
local attempt=1
log_step "Waiting for server $server_id to become active..."
while [[ "$attempt" -le "$max_attempts" ]]; do
local response
response=$(latitude_api GET "/servers/$server_id")
local status
status=$(echo "$response" | python3 -c "
import json, sys
data = json.loads(sys.stdin.read())
server = data.get('data', {})
attrs = server.get('attributes', {})
print(attrs.get('status', 'unknown'))
" 2>/dev/null || echo "unknown")
if [[ "$status" == "on" ]] || [[ "$status" == "active" ]]; then
LATITUDE_SERVER_IP=$(extract_latitude_server_ip "$response")
if [[ -n "$LATITUDE_SERVER_IP" ]]; then
export LATITUDE_SERVER_IP
log_info "Server active: IP=$LATITUDE_SERVER_IP"
return 0
fi
log_step "Server active but IP not yet assigned... (attempt $attempt/$max_attempts)"
else
log_step "Server status: $status (attempt $attempt/$max_attempts)"
fi
sleep 10
attempt=$((attempt + 1))
done
log_error "Server failed to become active after $max_attempts attempts"
return 1
INSTANCE_STATUS_POLL_DELAY=10 generic_wait_for_instance latitude_api \
"/servers/$server_id" \
"on" \
"d.get('data',{}).get('attributes',{}).get('status','unknown')" \
"${_LATITUDE_IP_PY}" \
LATITUDE_SERVER_IP "Latitude.sh server" "${max_attempts}"
}
# SSH operations — delegates to shared helpers (SSH_USER defaults to root)

View file

@ -115,11 +115,11 @@ ovh_register_ssh_key() {
import json, sys
pub_key = sys.stdin.read().strip()
body = {
'name': '$key_name',
'name': sys.argv[1],
'publicKey': pub_key
}
print(json.dumps(body))
")
" "$key_name")
local response
response=$(ovh_api_call POST "/cloud/project/${OVH_PROJECT_ID}/sshkey" "$body")
@ -178,13 +178,13 @@ _ovh_find_flavor_id() {
python3 -c "
import json, sys
flavors = json.loads(sys.stdin.read())
target = '${flavor_name}'
target = sys.argv[1]
for f in flavors:
if f.get('name', '') == target:
print(f['id'])
sys.exit(0)
print('')
" <<< "${flavors_response}"
" "$flavor_name" <<< "${flavors_response}"
}
# Get SSH key ID from OVH
@ -196,7 +196,7 @@ _ovh_get_ssh_key_id() {
python3 -c "
import json, sys
keys = json.loads(sys.stdin.read())
fp = '${fingerprint}'
fp = sys.argv[1]
for k in keys:
if fp in k.get('fingerprint', '') or fp in k.get('publicKey', ''):
print(k['id'])
@ -204,7 +204,7 @@ for k in keys:
# Fallback: return first key
if keys:
print(keys[0]['id'])
" <<< "${keys_response}"
" "$fingerprint" <<< "${keys_response}"
}
# Resolve image ID, flavor ID, and SSH key ID for OVH instance creation
@ -244,19 +244,18 @@ _ovh_resolve_resources() {
_ovh_build_instance_body() {
local name="$1" flavor_id="$2" image_id="$3" region="$4" ssh_key_id="$5"
python3 -c "
import json
import json, sys
body = {
'name': '${name}',
'flavorId': '${flavor_id}',
'imageId': '${image_id}',
'region': '${region}',
'name': sys.argv[1],
'flavorId': sys.argv[2],
'imageId': sys.argv[3],
'region': sys.argv[4],
'monthlyBilling': False
}
ssh_key_id = '${ssh_key_id}'
if ssh_key_id:
body['sshKeyId'] = ssh_key_id
if sys.argv[5]:
body['sshKeyId'] = sys.argv[5]
print(json.dumps(body))
"
" "$name" "$flavor_id" "$image_id" "$region" "$ssh_key_id"
}
# Create an OVH Public Cloud instance