Python 3 library and CLI application for SporeStack
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

288 lines
8.0 KiB

import logging
from time import sleep
from typing import Any, Dict, Optional, Union
import requests
from . import validate
log = logging.getLogger(__name__)
LATEST_API_VERSION = 2
CLEARNET_ENDPOINT = "https://api.sporestack.com"
TOR_ENDPOINT = "http://spore64i5sofqlfz5gq2ju4msgzojjwifls7rok2cti624zyq3fcelad.onion"
API_ENDPOINT = CLEARNET_ENDPOINT
GET_TIMEOUT = 60
POST_TIMEOUT = 90
USE_TOR_PROXY = "auto"
TOR_PROXY = "socks5h://127.0.0.1:9050"
# For requests module
TOR_PROXY_REQUESTS = {"http": TOR_PROXY, "https": TOR_PROXY}
def is_onion_url(url: str) -> bool:
"""
returns True/False depending on if a URL looks like a Tor hidden service
(.onion) or not.
This is designed to false as non-onion just to be on the safe-ish side,
depending on your view point. It requires URLs like: http://domain.tld/,
not http://domain.tld or domain.tld/.
This can be optimized a lot.
"""
try:
url_parts = url.split("/")
domain = url_parts[2]
tld = domain.split(".")[-1]
if tld == "onion":
return True
except Exception:
pass
return False
def api_request(
url: str,
json_params: Optional[Dict[str, Any]] = None,
get_params: Optional[Dict[str, Any]] = None,
retry: bool = False,
) -> Any:
# FIXME: Should return some request type
proxies = {}
if is_onion_url(url) is True:
log.debug("Got a .onion API endpoint, using local Tor SOCKS proxy.")
proxies = TOR_PROXY_REQUESTS
try:
if json_params is None:
request = requests.get(
url, params=get_params, timeout=GET_TIMEOUT, proxies=proxies
)
else:
request = requests.post(
url, json=json_params, timeout=POST_TIMEOUT, proxies=proxies
)
except Exception as e:
if retry is True:
log.warning(f"Got an error, but retrying: {e}")
sleep(5)
# Try again.
return api_request(
url,
json_params=json_params,
get_params=get_params,
retry=retry,
)
else:
raise
status_code_first_digit = request.status_code // 100
if status_code_first_digit == 2:
try:
request_dict = request.json()
if "latest_api_version" in request_dict:
if request_dict["latest_api_version"] > LATEST_API_VERSION:
log.warning("New API version may be available.")
return request_dict
except Exception:
return request.content
elif status_code_first_digit == 4:
if request.status_code == 415:
raise NotImplementedError(request.content.decode("utf-8"))
else:
log.debug("Status code: {request.status_code}")
raise ValueError(request.content.decode("utf-8"))
elif status_code_first_digit == 5:
if retry is True:
log.warning(request.content.decode("utf-8"))
log.warning("Got a 500, retrying in 5 seconds...")
sleep(5)
# Try again if we get a 500
return api_request(
url,
json_params=json_params,
get_params=get_params,
retry=retry,
)
else:
raise Exception(str(request.content))
else:
# Not sure why we'd get this.
request.raise_for_status()
raise Exception("Stuff broke strangely.")
def normalize_argument(argument: Union[str, None, bool]) -> Union[str, None, bool]:
"""
Helps normalize arguments from aaargh that may not be what we want.
"""
if argument == "False":
return False
elif argument == "True":
return True
elif argument == "None":
return None
else:
return argument
def launch(
machine_id: str,
days: int,
currency: str,
flavor: str,
operating_system: str,
ssh_key: str,
api_endpoint: str = API_ENDPOINT,
region: Optional[str] = None,
settlement_token: Optional[str] = None,
retry: bool = False,
affiliate_amount: Optional[int] = None,
affiliate_token: Optional[str] = None,
) -> Any:
validate.currency(currency)
validate.flavor(flavor)
validate.machine_id(machine_id)
validate.operating_system(operating_system)
validate.ssh_key(ssh_key)
validate.affiliate_amount(affiliate_amount)
json_params = {
"machine_id": machine_id,
"days": days,
"flavor": flavor,
"currency": currency,
"region": region,
"settlement_token": settlement_token,
"operating_system": operating_system,
"ssh_key": ssh_key,
"affiliate_amount": affiliate_amount,
"affiliate_token": affiliate_token,
}
url = f"{api_endpoint}/v2/launch"
return api_request(url=url, json_params=json_params, retry=retry)
def topup(
machine_id: str,
days: int,
currency: str,
api_endpoint: str = API_ENDPOINT,
settlement_token: Optional[str] = None,
retry: bool = False,
affiliate_amount: Optional[int] = None,
affiliate_token: Optional[str] = None,
) -> Any:
validate.machine_id(machine_id)
validate.currency(currency)
validate.affiliate_amount(affiliate_amount)
json_params = {
"machine_id": machine_id,
"days": days,
"settlement_token": settlement_token,
"currency": currency,
"affiliate_amount": affiliate_amount,
"affiliate_token": affiliate_token,
}
url = f"{api_endpoint}/v2/topup"
return api_request(url=url, json_params=json_params, retry=retry)
def start(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Boots the VM.
"""
validate.machine_id(machine_id)
url = f"{api_endpoint}/v2/start"
json_params = {"machine_id": machine_id}
api_request(url, json_params=json_params)
def stop(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Immediately kills the VM.
"""
validate.machine_id(machine_id)
url = f"{api_endpoint}/v2/stop"
json_params = {"machine_id": machine_id}
api_request(url, json_params=json_params)
def delete(machine_id: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Immediately deletes the VM.
"""
validate.machine_id(machine_id)
url = f"{api_endpoint}/v2/delete"
json_params = {"machine_id": machine_id}
api_request(url, json_params=json_params)
def info(machine_id: str, api_endpoint: str = API_ENDPOINT) -> Any:
"""
Returns info about the VM.
"""
validate.machine_id(machine_id)
url = f"{api_endpoint}/v2/info"
get_params = {"machine_id": machine_id}
return api_request(url, get_params=get_params)
def settlement_token_enable(
settlement_token: str,
cents: int,
currency: str,
api_endpoint: str = API_ENDPOINT,
retry: bool = False,
) -> Any:
validate.settlement_token(settlement_token)
validate.cents(cents)
validate.currency(currency)
json_params = {
"settlement_token": settlement_token,
"cents": cents,
"currency": currency,
}
url = api_endpoint + "/settlement/enable"
return api_request(url=url, json_params=json_params, retry=retry)
def settlement_token_add(
settlement_token: str,
cents: int,
currency: str,
api_endpoint: str = API_ENDPOINT,
retry: bool = False,
) -> Any:
validate.settlement_token(settlement_token)
validate.cents(cents)
validate.currency(currency)
json_params = {
"settlement_token": settlement_token,
"cents": cents,
"currency": currency,
}
url = api_endpoint + "/settlement/add"
return api_request(url=url, json_params=json_params, retry=retry)
def settlement_token_balance(
settlement_token: str, api_endpoint: str = API_ENDPOINT
) -> Any:
validate.settlement_token(settlement_token)
get_params = {"settlement_token": settlement_token}
url = api_endpoint + "/settlement/balance"
return api_request(url=url, get_params=get_params)