Rewrote ConfigGenerator in order to support more options

.gitignore

@@ -1,3 +1,5 @@
+/jwt
+/coverage-report
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

nats.conf

@@ -0,0 +1,18 @@
+port: 4222
+
+authorization: {
+  "password": "test",
+  "user": "test"
+}
+
+max_connections: 1
+
+websocket: {
+port: 8888
+
+tls {
+    "ca_file": "/tmp/nats_tools_tls_asuvvj4m-/ca.crt",
+    "cert_file": "/tmp/nats_tools_tls_asuvvj4m-/server.crt",
+    "key_file": "/tmp/nats_tools_tls_asuvvj4m-/server.key"
+}
+}

pyproject.toml

@@ -19,8 +19,10 @@ dynamic = ["version"]
 dependencies = ["httpx", "jinja2"]
 
 [project.optional-dependencies]
+tls = ["trustme"]
 build = ["build", "invoke", "pip-tools"]
 dev = [
+    "trustme",
     "black",
     "isort",
     "invoke",

src/nats_tools/__about__.py

@@ -16,4 +16,4 @@
 print(__version__)
 ```
 """
-__version__ = "1.0.2"
+__version__ = "2.0.0"

src/nats_tools/__init__.py

@@ -1,5 +1,11 @@
 from .__about__ import __version__
+from .config import ConfigGenerator, check_config
 from .natsd import NATSD, NATSMonitor
-from .templates import ConfigGenerator
 
-__all__ = ["__version__", "NATSD", "NATSMonitor", "ConfigGenerator"]
+__all__ = [
+    "__version__",
+    "NATSD",
+    "NATSMonitor",
+    "ConfigGenerator",
+    "check_config",
+]

src/nats_tools/cmd.py

@@ -0,0 +1,38 @@
+import os
+import shutil
+import subprocess
+import typing as t
+from pathlib import Path
+
+DEFAULT_BIN_DIR = Path.home().joinpath("nats-server").absolute()
+
+
+def get_executable(bin_name: str = "nats-server") -> str:
+    """Get path to nats-server executable"""
+    # User directory
+    if DEFAULT_BIN_DIR.joinpath(bin_name).is_file():
+        return DEFAULT_BIN_DIR.joinpath(bin_name).as_posix()
+    elif DEFAULT_BIN_DIR.joinpath(bin_name + ".exe").is_file():
+        return DEFAULT_BIN_DIR.joinpath(bin_name + ".exe").as_posix()
+    # Any directory within PATH
+    else:
+        path = shutil.which(bin_name)
+        if path is None:
+            raise FileNotFoundError("nats-server executable not found")
+        return path
+
+
+def nats_server(
+    *args: str,
+    stdout: t.Optional[t.Any] = None,
+    stderr: t.Optional[t.Any] = None,
+    max_cpus: t.Optional[int] = None,
+    executable: t.Optional[str] = None,
+) -> "subprocess.Popen[bytes]":
+    """Execute NATS server using subprocess.Popen"""
+    executable = executable or get_executable("nats-server")
+    args = (executable,) + args
+    env = os.environ.copy()
+    if max_cpus:
+        env["GOMAXPROCS"] = format(max_cpus, ".2f")
+    return subprocess.Popen(args, stdout=stdout, stderr=stderr, env=env)

src/nats_tools/config/__init__.py

@@ -0,0 +1,5 @@
+from .generator import ConfigGenerator, render
+from .options import ServerOptions
+from .utils import check_config, non_null
+
+__all__ = ["ServerOptions", "ConfigGenerator", "render", "non_null", "check_config"]

src/nats_tools/config/blocks/__init__.py

@@ -0,0 +1,39 @@
+from .accounts import (
+    Account,
+    AccountJetStreamLimits,
+    ServiceExport,
+    ServiceImport,
+    Source,
+    StreamExport,
+    StreamImport,
+)
+from .authorization import Authorization
+from .cluster import Cluster
+from .jetstream import JetStream
+from .leafnodes import LeafNodes, RemoteLeafnode
+from .mqtt import MQTT
+from .resolvers import NATSResolver
+from .tls import TLS
+from .users import Permissions, User
+from .websocket import Websocket
+
+__all__ = [
+    "Account",
+    "AccountJetStreamLimits",
+    "Authorization",
+    "Cluster",
+    "NATSResolver",
+    "JetStream",
+    "LeafNodes",
+    "MQTT",
+    "Permissions",
+    "RemoteLeafnode",
+    "ServiceExport",
+    "ServiceImport",
+    "Source",
+    "StreamExport",
+    "StreamImport",
+    "TLS",
+    "User",
+    "Websocket",
+]

src/nats_tools/config/blocks/accounts.py

@@ -0,0 +1,119 @@
+import typing as t
+from dataclasses import dataclass
+
+from .users import User
+
+
+@dataclass
+class StreamExport:
+    """Bind a subject for use as a stream from other accounts.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts#export-configuration-map
+    """
+
+    stream: str
+    """A subject or subject with wildcards that the account will publish."""
+
+    accounts: t.Optional[t.List[str]] = None
+    """A list of account names that can import the stream. If not specified, the stream is public and any account can import it."""
+
+
+@dataclass
+class ServiceExport:
+    """Bind a subject for use as a service from other accounts.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts#export-configuration-map
+    """
+
+    service: str
+    """A subject or subject with wildcards that the account will subscribe to."""
+
+    accounts: t.Optional[t.List[str]] = None
+    """A list of account names that can import the service. If not specified, the service is public and any account can import it."""
+
+    response_type: t.Optional[str] = None
+    """Indicates if a response to a service request consists of a single or a stream of messages.
+    Possible values are: single or stream. (Default value is singleton)
+    """
+
+
+@dataclass
+class Source:
+    """Source configuration map.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts#source-configuration-map
+    """
+
+    account: str
+    """Account name owning the export."""
+
+    subject: str
+    """The subject under which the stream or service is made accessible to the importing account"""
+
+
+@dataclass
+class StreamImport:
+    """Enables account to consume a stream published by another account.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts#import-configuration-map
+    """
+
+    stream: Source
+    """Stream import source configuration."""
+
+    prefix: t.Optional[str] = None
+    """A local subject prefix mapping for the imported stream."""
+
+
+@dataclass
+class ServiceImport:
+    """Enables account to consume a service implemented by another account.
+
+    References: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts#import-configuration-map
+    """
+
+    service: Source
+    """Service import source configuration."""
+
+    to: t.Optional[str] = None
+    """A local subject mapping for imported service."""
+
+
+@dataclass
+class AccountJetStreamLimits:
+    """JetStream account limits.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/resource_management#setting-account-resource-limits.
+    """
+
+    max_mem: t.Union[str, int, None] = None
+    """Maximum size of the `memory` storage. Default unit is bytes when integer is provided."""
+
+    max_file: t.Union[str, int, None] = None
+    """Maximum size of the `file` storage. Default unit is bytes when integer is provided."""
+
+    max_streams: t.Optional[int] = None
+    """Maximum number of streams."""
+
+    max_consumers: t.Optional[int] = None
+    """Maximum number of consumers."""
+
+
+@dataclass
+class Account:
+    """Multi-tenance account configuration.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/accounts
+    """
+
+    users: t.List[User]
+    """Users which can connect to the account using username/password auth."""
+
+    exports: t.Optional[t.Sequence[t.Union[ServiceExport, StreamExport]]] = None
+    """Define account exports."""
+
+    imports: t.Optional[t.Sequence[t.Union[StreamImport, ServiceImport]]] = None
+    """Define account imports."""
+
+    jetstream: t.Optional[AccountJetStreamLimits] = None
+    """Resource limits for the account."""

src/nats_tools/config/blocks/authorization.py

@@ -0,0 +1,24 @@
+import typing as t
+from dataclasses import dataclass
+
+from .users import User
+
+
+@dataclass
+class Authorization:
+    """Client Authentication/Authorization
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro#authorization-map
+    """
+
+    user: t.Optional[str] = None
+    """Specifies a single global user name for clients to the server (exclusive of token)"""
+
+    password: t.Optional[str] = None
+    """Specifies a single global password for clients to the server (exclusive of token)."""
+
+    token: t.Optional[str] = None
+    """Specifies a global token that can be used to authenticate to the server (exclusive of user and password)"""
+
+    users: t.Optional[t.List[User]] = None
+    """A list of user configuration maps. For multiple username and password credentials, specify a users list."""

src/nats_tools/config/blocks/cluster.py

@@ -0,0 +1,47 @@
+import typing as t
+from dataclasses import dataclass
+
+from .authorization import Authorization
+from .tls import TLS
+
+
+@dataclass
+class Cluster:
+    """NATS cluster mode configuration.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration/clustering/cluster_config
+    """
+
+    host: t.Optional[str] = None
+    """Interface where the gateway will listen for incoming route connections."""
+
+    port: t.Optional[int] = None
+    """Port where the gateway will listen for incoming route connections."""
+
+    listen: t.Optional[str] = None
+    """Combines host and port as `<host>:<port>`."""
+
+    tls: t.Optional[TLS] = None
+    """A tls configuration map for securing the clustering connection. verify is always enabled and cert_file is used for client and server."""
+
+    name: t.Optional[str] = None
+    """Name of the cluster."""
+
+    advertise: t.Optional[str] = None
+    """Hostport <host>:<port> to advertise how this server can be contacted by other cluster members."""
+
+    no_advertise: t.Optional[bool] = None
+    """When set to 'true', the server will not send or gossip its client URLs to other servers in the cluster and will not tell its client about the other servers' client URLs."""
+
+    routes: t.Optional[t.List[str]] = None
+    """A list of other servers (URLs) to cluster with. Self-routes are ignored. Should authentication via token or username/password be required, specify them as part of the URL."""
+
+    connect_retries: t.Optional[int] = None
+    """After how many failed connect attempts to give up establishing a connection to a discovered route. Default is 0, do not retry. When enabled, attempts will be made once a second. This, does not apply to explicitly configured routes."""
+
+    authorization: t.Optional[Authorization] = None
+    """Authorization map for configuring cluster routes.
+    When a single username/password is used, it defines the authentication mechanism this server expects,
+    and how this server will authenticate itself when establishing a connection to a discovered route.
+    This will not be used for routes explicitly listed in routes and therefore have to be provided as
+    part of the URL."""

src/nats_tools/config/blocks/jetstream.py

@@ -0,0 +1,32 @@
+import typing as t
+from dataclasses import dataclass
+
+
+@dataclass
+class JetStream:
+    """Enable and configure JetStream.
+
+    Reference: https://docs.nats.io/running-a-nats-service/configuration#jetstream
+    """
+
+    store_dir: t.Optional[str] = None
+    """Directory to use for JetStream storage.
+    Default to `/tmp/nats/jetstream`."""
+
+    max_mem: t.Union[str, int, None] = None
+    """Maximum size of the `memory` storage. Default to `75%` of available memory."""
+
+    max_file: t.Union[str, int, None] = None
+    """Maximum size of the `file` storage. Up to `1TB` if available."""
+
+    cipher: t.Optional[str] = None
+    """Set to enable storage-level encryption at rest. Choose either `chachapoly` or `aes`"""
+
+    key: t.Optional[str] = None
+    """The encryption key to use when encryption is enabled. A key length of at least 32 bytes is recommended. Note, this key is HMAC-256 hashed on startup which reduces the byte length to 64."""
+
+    max_outstanding_catchup: t.Optional[str] = None
+    """Max in-flight bytes for stream catch-up. Default to `32MB`."""
+
+    domain: t.Optional[str] = None
+    """Jetstream domain."""