from datetime import datetime, timedelta
from typing import List, Union, cast, Optional, TYPE_CHECKING
import dateutil.parser
from . import htb, vpn
from .errors import IncorrectArgumentException, IncorrectFlagException, TooManyResetAttempts, MachineException
from .solve import MachineSolve
from .utils import parse_delta
if TYPE_CHECKING:
from .user import User
[docs]class Machine(htb.HTBObject):
""" The class representing Hack The Box machines
Attributes:
name: The Machine name
os: The name of the operating system
points: The points awarded for completion
release_date: The date the Machine was released
user_owns: The number of user owns the Machine has
root_owns: The number of root owns the Machine has
free: Whether the Machine is available on free servers
user_owned: Whether the active User has owned the Machine's user account
root_owned: Whether the active User has owned the Machine's user account
reviewed: Whether the active User has reviewed the Machine
stars: The average star rating of the Machine
avatar: The relative URL of the Machine avatar
difficulty: The difficulty of the machine
:noindex: ip: The IP address of the machine
active: Whether the Machine is active
retired: Whether the Machine is retired
avg_difficulty: The average numeric difficulty of the Machine
completed: Whether the active User has completed the Machine
:noindex: user_own_time: How long the active User took to own user
:noindex: root_own_time: How long the active User took to own root
user_blood: The Solve of the Machine's first user blood
root_blood: The Solve of the Machine's first root blood
user_own_time: How long the first User took to own user
root_own_time: How long the first User took to own root
difficulty_ratings: A dict of difficulty ratings given
"""
name: str
os: str
points: int
release_date: datetime
root_owns: int
free: bool
user_owned: bool
root_owned: bool
reviewed: bool
stars: float
avatar: str
difficulty: str
_detailed_attributes = ('active', 'retired', 'user_own_time', 'root_own_time', 'user_blood',
'root_blood', 'user_blood_time', 'root_blood_time', 'difficulty_ratings')
active: bool
retired: bool
avg_difficulty: int
completed: bool
user_own_time: timedelta
root_own_time: timedelta
user_blood: MachineSolve
root_blood: MachineSolve
user_blood_time: timedelta
root_blood_time: timedelta
difficulty_ratings: dict
# noinspection PyUnresolvedReferences
_authors: Optional[List["User"]] = None
_author_ids: List[int]
_is_release: Optional[bool] = None
_ip: Optional[str] = None
[docs] def submit(self, flag: str, difficulty: int):
""" Submits a flag for a Machine
Args:
flag: The flag for the Machine
difficulty: A rating between 10 and 100 of the Machine difficulty.
Must be a multiple of 10.
"""
if difficulty < 10 or difficulty > 100 or difficulty % 10 != 0:
raise IncorrectArgumentException(reason="Difficulty must be a multiple of 10, between 10 and 100")
submission = cast(dict, self._client.do_request("machine/own", json_data={
"flag": flag,
"id": self.id,
"difficulty": difficulty
}))
if submission['message'] == "Incorrect flag!":
raise IncorrectFlagException
return True
# noinspection PyUnresolvedReferences
@property
def authors(self) -> List["User"]:
"""Fetch the author(s) of the Machine
Returns: List of Users
"""
if not self._authors:
self._authors = []
for uid in self._author_ids:
self._authors.append(self._client.get_user(uid))
return self._authors
@property
def is_release(self):
if self._is_release is not None:
return self._is_release
self._is_release = False
data = self._client.do_request("connections")["data"]
try:
if data['release_arena']['machine']['id'] == self.id:
self._is_release = True
except AttributeError:
pass
return self._is_release
@property
def ip(self):
"""The IP of an active machine."""
if self._ip is not None:
return self._ip
listing = self._client.do_request("machine/list")["info"]
m = next(filter(lambda x: x["id"] == self.id, listing))
self._ip = m["ip"]
return self._ip
[docs] def start(self, release_arena=False) -> Union["MachineInstance", None]:
"""Alias for `Machine.spawn()`"""
return self.spawn(release_arena)
[docs] def spawn(self, release_arena=False) -> "MachineInstance":
"""Spawn an instance of this machine.
Args:
release_arena: Whether to use Release Arena to spawn the machine
Returns:
The spawned `MachineInstance`
"""
if release_arena:
if not self.is_release:
# TODO: Better exception
raise Exception("Machine is not on release arena")
data = cast(dict, self._client.do_request("release_arena/spawn", post=True))
if data.get("success") != 1:
raise Exception(f"Failed to spawn: {data}")
ip = cast(dict, self._client.do_request("release_arena/active"))["info"]["ip"]
server = self._client.get_current_vpn_server(release_arena=True)
else:
data = cast(dict, self._client.do_request("vm/spawn", json_data={"machine_id": self.id}))
if "Machine deployed" in cast(str, data.get("message")) or "You have been assigned" in cast(str, data.get("message")):
ip = cast(dict, self._client.do_request(f"machine/profile/{self.id}"))["info"]["ip"]
server = self._client.get_current_vpn_server()
else:
raise Exception(f"Failed to spawn: {data}")
return MachineInstance(ip, server, self, self._client)
def __repr__(self):
return f"<Machine '{self.name}'>"
def __init__(self, data: dict, client: htb.HTBClient, summary: bool = False):
self._client = client
self._detailed_func = client.get_machine # type: ignore
self.id = data['id']
self.name = data['name']
self.os = data['os']
self.points = data['points']
self.release_date = dateutil.parser.parse(data['release'])
self.user_owns = data['user_owns_count']
self.root_owns = data['root_owns_count']
self.user_owned = data['authUserInUserOwns']
self.root_owned = data['authUserInRootOwns']
self.reviewed = data['authUserHasReviewed']
self.stars = float(data['stars'])
self.avatar = data['avatar']
self.difficulty = data['difficultyText']
self.free = data['free']
self._author_ids = [data['maker']['id']]
if data.get('ip'):
self._ip = data['ip']
if data['maker2']:
self._author_ids.append(data['maker2']['id'])
if not summary:
self.active = bool(data['active'])
self.retired = bool(data['retired'])
if data['authUserInUserOwns']:
self.user_own_time = parse_delta(data['authUserFirstUserTime'])
if data['authUserInRootOwns']:
self.root_own_time = parse_delta(data['authUserFirstRootTime'])
self.difficulty_ratings = data['feedbackForChart']
if data['userBlood']:
user_blood_data = {
"date": dateutil.parser.parse(data['userBlood']['created_at']),
"first_blood": True,
"id": data['id'],
"name": data['name'],
"type": "user"
}
self.user_blood = MachineSolve(user_blood_data, self._client)
self.user_blood_time = parse_delta(data['userBlood']['blood_difference'])
if data['rootBlood']:
user_blood_data = {
"date": dateutil.parser.parse(data['rootBlood']['created_at']),
"first_blood": True,
"id": data['id'],
"name": data['name'],
"type": "root"
}
self.root_blood = MachineSolve(user_blood_data, self._client)
self.root_blood_time = parse_delta(data['rootBlood']['blood_difference'])
else:
self._is_summary = True
[docs]class MachineInstance:
"""Representation of an active machine instance
Attributes:
ip: The IP the instance can be reached at
server: The `VPNServer` that the machine is on
machine: The `Machine` this is an instance of
client: The passed-through API client
"""
ip: str
server: vpn.VPNServer
client: htb.HTBClient
machine: Machine
def __init__(self, ip: str, server: vpn.VPNServer, machine: Machine, client: htb.HTBClient):
self.client = client
self.ip = ip
self.server = server
self.machine = machine
def __repr__(self):
return f"<'{self.machine.name}'@{self.server.friendly_name} - {self.ip}>"
[docs] def stop(self):
"""Request the instance be stopped."""
if self.machine.is_release:
self.client.do_request("release_arena/terminate", post=True)
else:
self.client.do_request("vm/terminate", json_data={"machine_id": self.machine.id})
# Can't delete references to the object from here so we just have
# to set everything to None and prevent further usage
self.server = None
self.ip = None
self.client = None
self.machine = None
[docs] def reset(self):
"""Request the instance be reset."""
if self.machine.is_release:
resp = self.client.do_request("release_arena/reset", json_data={"machine_id": self.machine.id})
else:
resp = self.client.do_request("vm/reset", json_data={"machine_id": self.machine.id})
# VM
if resp["message"].endswith(" will be reset in 1 minute."):
return True
elif resp["message"] == "Too many reset machine attempts. Try again later!":
raise TooManyResetAttempts
# RA
elif resp["success"]:
return True
elif resp["message"].startswith("You must wait"):
raise TooManyResetAttempts
raise MachineException