First working version

This commit is contained in:
2024-06-27 16:31:55 +02:00
parent df5754518d
commit 737716f8eb
57 changed files with 1508 additions and 0 deletions

View File

@ -0,0 +1,4 @@
from .apikeys import APIKey
from .nodes import Node
from .users import User
from .routes import Route

View File

@ -0,0 +1,26 @@
from .model import HSAPICall
from headscale_api_client.schemas import (v1CreateApiKeyRequest,
v1ListApiKeysResponse,
v1CreateApiKeyResponse,
v1ExpireApiKeyRequest,
v1ExpireApiKeyResponse,
v1DeleteApiKeyResponse)
class APIKey(HSAPICall):
def list(self) -> v1ListApiKeysResponse:
response = self.call('get', 'apikey')
return v1ListApiKeysResponse(**response.json())
def create(self, data: v1CreateApiKeyRequest) -> v1CreateApiKeyResponse:
response = self.call('post', 'apikey', data)
return v1CreateApiKeyResponse(**response.json())
def expire(self, data: v1ExpireApiKeyRequest) -> v1ExpireApiKeyResponse:
response = self.call('post', 'apikey/expire', data)
return v1ExpireApiKeyResponse(**response.json())
def delete(self, prefix: str) -> v1DeleteApiKeyResponse:
response = self.call('delete', f'apikey/{prefix}')
return v1DeleteApiKeyResponse(**response.json())

View File

@ -0,0 +1,25 @@
import os
from typing import Optional, Union
from pydantic_settings import BaseSettings, SettingsConfigDict
class APISettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='HS_')
server: str = "http://localhost:8080"
api_path: str = "/api/v1"
ssl_verify: Union[bool, str] = True
api_token: Union[None, str] = None
def refresh_api_token(self):
self.api_token = os.environ.get('HS_API_TOKEN')
class HTTPException(Exception):
def __init__(self, status_code: int, message: str):
self.status_code = status_code
self.message = message
super().__init__(f"{status_code} {message}")
def __str__(self):
return f"{self.status_code} {self.message}"

View File

@ -0,0 +1,48 @@
from .schemas import *
from .config import APISettings, HTTPException
import requests
class HSAPICall:
"""
Generic API call.
It has a call() method that wants:
- a method (GET, POST, DELETE);
- a subpath, that is appended to <server>/api/v1
- optional `data` payload
"""
objectPath: str = ""
def __init__(self, api_settings_override: Optional[APISettings] = None) -> None:
if api_settings_override:
self.api_settings = api_settings_override
else:
self.api_settings = APISettings()
self.base_path = f"{
self.api_settings.server}{self.api_settings.api_path}/{self.objectPath}"
def call(self, method, call_path: str = "", data=None):
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": f"Bearer {self.api_settings.api_token}",
}
json_ = data.dict() if data else dict()
path = '/'.join([self.base_path, str(call_path)]
) if call_path else self.base_path
response = requests.request(
method,
path,
headers=headers,
verify=self.api_settings.ssl_verify,
json=json_
)
if response.status_code != 200:
raise HTTPException(response.status_code, f" failed with status code: {
response.text}")
return response

View File

@ -0,0 +1,48 @@
from .model import HSAPICall
from headscale_api_client.schemas import (v1ListNodesResponse,
v1GetNodeResponse,
v1DeleteNodeResponse,
v1ExpireNodeResponse, v1MoveNodeRequest,
v1RenameNodeResponse,
v1MoveNodeRequest,
v1MoveNodeResponse,
v1GetNodeRoutesResponse,
v1SetTagsResponse,
HeadscaleServiceSetTagsBody)
class Node(HSAPICall):
objectPath = "node"
def list(self) -> v1ListNodesResponse:
response = self.call('get')
return v1ListNodesResponse(**response.json())
def get(self, nodeId: str) -> v1GetNodeResponse:
response = self.call('get', call_path=nodeId)
return v1GetNodeResponse(**response.json())
def delete(self, nodeId: str) -> v1DeleteNodeResponse:
response = self.call('delete', call_path=nodeId)
return v1DeleteNodeResponse(**response.json())
def expire(self, nodeId: str) -> v1ExpireNodeResponse:
response = self.call('post', f'{nodeId}/expire')
return v1ExpireNodeResponse()(**response.json())
def rename(self, nodeId: str, newName: str) -> v1RenameNodeResponse:
response = self.call('post', f'{nodeId}/rename/{newName}')
return v1RenameNodeResponse(**response.json())
def move(self, nodeId: str, data: v1MoveNodeRequest) -> v1MoveNodeResponse:
response = self.call('post', f'{nodeId}/user', data)
return v1MoveNodeResponse()(**response.json())
def routes(self, nodeId: str) -> v1GetNodeRoutesResponse:
response = self.call('get', f'{nodeId}/routes')
return v1GetNodeRoutesResponse(**response.json())
def setTags(self, nodeId: str, data: HeadscaleServiceSetTagsBody) -> v1SetTagsResponse:
response = self.call('post', f'{nodeId}/tags', data)
return v1SetTagsResponse(**response.json())

View File

@ -0,0 +1,27 @@
from .model import HSAPICall
from headscale_api_client.schemas import (v1GetRoutesResponse,
v1DeleteRouteResponse,
v1EnableRouteResponse,
v1DisableRouteResponse,
)
class Route(HSAPICall):
objectPath = "routes"
def list(self) -> v1GetRoutesResponse:
response = self.call('get')
return v1GetRoutesResponse(**response.json())
def delete(self, routeId: str) -> v1DeleteRouteResponse:
response = self.call('delete', call_path=routeId)
return v1DeleteRouteResponse(**response.json())
def enable(self, routeId: str) -> v1EnableRouteResponse:
response = self.call('post', f'{routeId}/enable')
return v1EnableRouteResponse(**response.json())
def disable(self, routeId: str) -> v1DisableRouteResponse:
response = self.call('post', f'{routeId}/disable')
return v1DisableRouteResponse(**response.json())

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class HeadscaleServiceSetTagsBody(BaseModel):
"""
None model
"""
tags: Optional[List[str]] = Field(alias="tags", default=None)

View File

@ -0,0 +1,42 @@
from .HeadscaleServiceSetTagsBody import *
from .protobufAny import *
from .rpcStatus import *
from .v1ApiKey import *
from .v1BackfillNodeIPsResponse import *
from .v1CreateApiKeyRequest import *
from .v1CreateApiKeyResponse import *
from .v1CreatePreAuthKeyRequest import *
from .v1CreatePreAuthKeyResponse import *
from .v1CreateUserRequest import *
from .v1CreateUserResponse import *
from .v1DebugCreateNodeRequest import *
from .v1DebugCreateNodeResponse import *
from .v1DeleteApiKeyResponse import *
from .v1DeleteNodeResponse import *
from .v1DeleteRouteResponse import *
from .v1DeleteUserResponse import *
from .v1DisableRouteResponse import *
from .v1EnableRouteResponse import *
from .v1ExpireApiKeyRequest import *
from .v1ExpireApiKeyResponse import *
from .v1ExpireNodeResponse import *
from .v1ExpirePreAuthKeyRequest import *
from .v1ExpirePreAuthKeyResponse import *
from .v1GetNodeResponse import *
from .v1GetNodeRoutesResponse import *
from .v1GetRoutesResponse import *
from .v1GetUserResponse import *
from .v1ListApiKeysResponse import *
from .v1ListNodesResponse import *
from .v1ListPreAuthKeysResponse import *
from .v1ListUsersResponse import *
from .v1MoveNodeResponse import *
from .v1Node import *
from .v1PreAuthKey import *
from .v1RegisterMethod import *
from .v1RegisterNodeResponse import *
from .v1RenameNodeResponse import *
from .v1RenameUserResponse import *
from .v1Route import *
from .v1SetTagsResponse import *
from .v1User import *

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class protobufAny(BaseModel):
"""
None model
"""
type: Optional[str] = Field(alias="@type", default=None)

View File

@ -0,0 +1,18 @@
from typing import *
from pydantic import BaseModel, Field
from .protobufAny import protobufAny
class rpcStatus(BaseModel):
"""
None model
"""
code: Optional[int] = Field(alias="code", default=None)
message: Optional[str] = Field(alias="message", default=None)
details: Optional[List[Optional[protobufAny]]] = Field(alias="details", default=None)

View File

@ -0,0 +1,20 @@
from typing import *
from pydantic import BaseModel, Field
class v1ApiKey(BaseModel):
"""
None model
"""
id: Optional[str] = Field(alias="id", default=None)
prefix: Optional[str] = Field(alias="prefix", default=None)
expiration: Optional[str] = Field(alias="expiration", default=None)
createdAt: Optional[str] = Field(alias="createdAt", default=None)
lastSeen: Optional[str] = Field(alias="lastSeen", default=None)

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class v1BackfillNodeIPsResponse(BaseModel):
"""
None model
"""
changes: Optional[List[str]] = Field(alias="changes", default=None)

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class v1CreateApiKeyRequest(BaseModel):
"""
None model
"""
expiration: Optional[str] = Field(alias="expiration", default=None)

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class v1CreateApiKeyResponse(BaseModel):
"""
None model
"""
apiKey: Optional[str] = Field(alias="apiKey", default=None)

View File

@ -0,0 +1,20 @@
from typing import *
from pydantic import BaseModel, Field
class v1CreatePreAuthKeyRequest(BaseModel):
"""
None model
"""
user: Optional[str] = Field(alias="user", default=None)
reusable: Optional[bool] = Field(alias="reusable", default=None)
ephemeral: Optional[bool] = Field(alias="ephemeral", default=None)
expiration: Optional[str] = Field(alias="expiration", default=None)
aclTags: Optional[List[str]] = Field(alias="aclTags", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1PreAuthKey import v1PreAuthKey
class v1CreatePreAuthKeyResponse(BaseModel):
"""
None model
"""
preAuthKey: Optional[v1PreAuthKey] = Field(alias="preAuthKey", default=None)

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class v1CreateUserRequest(BaseModel):
"""
None model
"""
name: Optional[str] = Field(alias="name", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1User import v1User
class v1CreateUserResponse(BaseModel):
"""
None model
"""
user: Optional[v1User] = Field(alias="user", default=None)

View File

@ -0,0 +1,18 @@
from typing import *
from pydantic import BaseModel, Field
class v1DebugCreateNodeRequest(BaseModel):
"""
None model
"""
user: Optional[str] = Field(alias="user", default=None)
key: Optional[str] = Field(alias="key", default=None)
name: Optional[str] = Field(alias="name", default=None)
routes: Optional[List[str]] = Field(alias="routes", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1DebugCreateNodeResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1DeleteApiKeyResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1DeleteNodeResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1DeleteRouteResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1DeleteUserResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1DisableRouteResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1EnableRouteResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class v1ExpireApiKeyRequest(BaseModel):
"""
None model
"""
prefix: Optional[str] = Field(alias="prefix", default=None)

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1ExpireApiKeyResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1ExpireNodeResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
class v1ExpirePreAuthKeyRequest(BaseModel):
"""
None model
"""
user: Optional[str] = Field(alias="user", default=None)
key: Optional[str] = Field(alias="key", default=None)

View File

@ -0,0 +1,10 @@
from typing import *
from pydantic import BaseModel, Field
class v1ExpirePreAuthKeyResponse(BaseModel):
"""
None model
"""

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1GetNodeResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Route import v1Route
class v1GetNodeRoutesResponse(BaseModel):
"""
None model
"""
routes: Optional[List[Optional[v1Route]]] = Field(alias="routes", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Route import v1Route
class v1GetRoutesResponse(BaseModel):
"""
None model
"""
routes: Optional[List[Optional[v1Route]]] = Field(alias="routes", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1User import v1User
class v1GetUserResponse(BaseModel):
"""
None model
"""
user: Optional[v1User] = Field(alias="user", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1ApiKey import v1ApiKey
class v1ListApiKeysResponse(BaseModel):
"""
None model
"""
apiKeys: Optional[List[Optional[v1ApiKey]]] = Field(alias="apiKeys", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1ListNodesResponse(BaseModel):
"""
None model
"""
nodes: Optional[List[Optional[v1Node]]] = Field(alias="nodes", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1PreAuthKey import v1PreAuthKey
class v1ListPreAuthKeysResponse(BaseModel):
"""
None model
"""
preAuthKeys: Optional[List[Optional[v1PreAuthKey]]] = Field(alias="preAuthKeys", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1User import v1User
class v1ListUsersResponse(BaseModel):
"""
None model
"""
users: Optional[List[Optional[v1User]]] = Field(alias="users", default=None)

View File

@ -0,0 +1,12 @@
from typing import *
from pydantic import BaseModel, Field
class v1DebugCreateNodeRequest(BaseModel):
"""
None model
"""
user: Optional[str] = Field(alias="user", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1MoveNodeResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,36 @@
from typing import *
from pydantic import BaseModel, Field
from .v1PreAuthKey import v1PreAuthKey
from .v1RegisterMethod import v1RegisterMethod
from .v1User import v1User
class v1Node(BaseModel):
"""
None model
"""
id: Optional[str] = Field(alias="id", default=None)
machineKey: Optional[str] = Field(alias="machineKey", default=None)
nodeKey: Optional[str] = Field(alias="nodeKey", default=None)
discoKey: Optional[str] = Field(alias="discoKey", default=None)
ipAddresses: Optional[List[str]] = Field(alias="ipAddresses", default=None)
name: Optional[str] = Field(alias="name", default=None)
user: Optional[v1User] = Field(alias="user", default=None)
lastSeen: Optional[str] = Field(alias="lastSeen", default=None)
expiry: Optional[str] = Field(alias="expiry", default=None)
preAuthKey: Optional[v1PreAuthKey] = Field(
alias="preAuthKey", default=None)
createdAt: Optional[str] = Field(alias="createdAt", default=None)
registerMethod: Optional[v1RegisterMethod] = Field(
alias="registerMethod", default=None)
forcedTags: Optional[List[str]] = Field(alias="forcedTags", default=None)
invalidTags: Optional[List[str]] = Field(alias="invalidTags", default=None)
validTags: Optional[List[str]] = Field(alias="validTags", default=None)
givenName: Optional[str] = Field(alias="givenName", default=None)
online: Optional[bool] = Field(alias="online", default=None)

View File

@ -0,0 +1,28 @@
from typing import *
from pydantic import BaseModel, Field
class v1PreAuthKey(BaseModel):
"""
None model
"""
user: Optional[str] = Field(alias="user", default=None)
id: Optional[str] = Field(alias="id", default=None)
key: Optional[str] = Field(alias="key", default=None)
reusable: Optional[bool] = Field(alias="reusable", default=None)
ephemeral: Optional[bool] = Field(alias="ephemeral", default=None)
used: Optional[bool] = Field(alias="used", default=None)
expiration: Optional[str] = Field(alias="expiration", default=None)
createdAt: Optional[str] = Field(alias="createdAt", default=None)
aclTags: Optional[List[str]] = Field(alias="aclTags", default=None)

View File

@ -0,0 +1,9 @@
from enum import Enum
class v1RegisterMethod(str, Enum):
REGISTER_METHOD_UNSPECIFIED = "REGISTER_METHOD_UNSPECIFIED"
REGISTER_METHOD_AUTH_KEY = "REGISTER_METHOD_AUTH_KEY"
REGISTER_METHOD_CLI = "REGISTER_METHOD_CLI"
REGISTER_METHOD_OIDC = "REGISTER_METHOD_OIDC"

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1RegisterNodeResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1RenameNodeResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1User import v1User
class v1RenameUserResponse(BaseModel):
"""
None model
"""
user: Optional[v1User] = Field(alias="user", default=None)

View File

@ -0,0 +1,30 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1Route(BaseModel):
"""
None model
"""
id: Optional[str] = Field(alias="id", default=None)
node: Optional[v1Node] = Field(alias="node", default=None)
prefix: Optional[str] = Field(alias="prefix", default=None)
advertised: Optional[bool] = Field(alias="advertised", default=None)
enabled: Optional[bool] = Field(alias="enabled", default=None)
isPrimary: Optional[bool] = Field(alias="isPrimary", default=None)
createdAt: Optional[str] = Field(alias="createdAt", default=None)
updatedAt: Optional[str] = Field(alias="updatedAt", default=None)
deletedAt: Optional[str] = Field(alias="deletedAt", default=None)

View File

@ -0,0 +1,14 @@
from typing import *
from pydantic import BaseModel, Field
from .v1Node import v1Node
class v1SetTagsResponse(BaseModel):
"""
None model
"""
node: Optional[v1Node] = Field(alias="node", default=None)

View File

@ -0,0 +1,16 @@
from typing import *
from pydantic import BaseModel, Field
class v1User(BaseModel):
"""
None model
"""
id: Optional[str] = Field(alias="id", default=None)
name: Optional[str] = Field(alias="name", default=None)
createdAt: Optional[str] = Field(alias="createdAt", default=None)

View File

@ -0,0 +1,33 @@
from .model import HSAPICall
from headscale_api_client.schemas import (v1CreateUserRequest,
v1CreateUserResponse,
v1DeleteUserResponse,
v1ListUsersResponse,
v1GetUserResponse,
v1RenameUserResponse,
)
class User(HSAPICall):
objectPath = "user"
def list(self) -> v1ListUsersResponse:
response = self.call('get')
return v1ListUsersResponse(**response.json())
def get(self, name: str) -> v1GetUserResponse:
response = self.call('get', call_path=name)
return v1GetUserResponse(**response.json())
def create(self, data: v1CreateUserRequest) -> v1CreateUserResponse:
response = self.call('post', data=data)
return v1CreateUserResponse(**response.json())
def delete(self, name: str) -> v1DeleteUserResponse:
response = self.call('delete', name)
return v1DeleteUserResponse(**response.json())
def rename(self, oldName: str, newName: str) -> v1RenameUserResponse:
response = self.call('post', call_path=f"{oldName}/rename/{newName}")
return v1RenameUserResponse(**response.json())