renamed cleanup_services.py

README.md

@@ -0,0 +1,45 @@
+# AppDaemon VSCode Tunnel
+
+## Install Service
+
+Use the `install_service.py` script to install.
+
+- Creates symlinks to `/etc/systemd/system`
+    - `example.socket`
+    - `example.service`
+- Reloads the service definitions
+- Starts the `example.socket`
+- Enables the `example.socket` to start at boot
+
+> [!NOTE]  
+> The `example.socket` does cause `example.service` to start when something connects to the socket at `/run/example.sock`.
+
+> [!CAUTION]
+> `example.service` does not stop when disconnecting from `/run/example.sock`
+
+## Scripts
+
+- `install_service.py`: Installs the services on the system
+- `run_service.py`: Used to demo running the code tunnel from a python function
+- `test_socket.py`: Opens a connection to the socket
+- `cleanup_services.sh`: Stops services and removes symlinks
+
+## Other Commands
+
+Reload systemd units
+
+```shell
+sudo systemctl daemon-reload
+```
+
+List loaded units
+
+```shell
+systemctl list-units --type=service
+```
+
+Check statuses
+
+```shell
+systemctl status example.socket
+```

cleanup_services.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python
+
+# for service in $(systemctl list-units --type=service --no-legend | grep example | awk '{print $1}'); do
+#     sudo systemctl stop $service
+#     echo "Stopped $service"
+# done
+
+import subprocess
+from pathlib import Path
+from typing import List
+
+
+def list_units() -> List[str]:
+    result = subprocess.run(
+        ['systemctl', 'list-units', '--type=service', '--no-legend'],
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+    return [line.split()[0] for line in result.stdout.splitlines()]
+
+
+def stop_service(service: str):
+    result = subprocess.run(['sudo', 'systemctl', 'stop', service])
+    if result.returncode == 0:
+        print(f'Stopped {service}')
+    return result.returncode
+
+
+def cleanup_service(service: str):
+    stop_result = stop_service(service)
+
+    service_file = Path('/etc/systemd/system') / service
+    if service_file.exists():
+        service_file.unlink()
+
+    if stop_result == 0:
+        subprocess.run(['sudo', 'systemctl', 'daemon-reload'], check=True)
+        print('Reloaded services')
+
+
+if __name__ == '__main__':
+    names = list_units()
+    for service in ['example.service', 'example.socket']:
+        cleanup_service(service)

cleanup_services.sh

@@ -1,6 +0,0 @@
-#!/bin/bash
-
-for service in $(systemctl list-units --type=service --no-legend | grep example | awk '{print $1}'); do
-    sudo systemctl stop $service
-    echo "Stopped $service"
-done

install_service.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python
 
 import subprocess
 from pathlib import Path
@@ -21,34 +21,32 @@ def create_symlink(source, target):
         print(f'Error creating symlink: [bold red]{e}[/]')
 
 
-def run_command(command):
-    try:
-        subprocess.run(command, check=True, shell=True)
-    except subprocess.CalledProcessError as e:
-        print(f'Error running command: {command}\n{e}')
-
-
 def main():
     repo_dir = Path(__file__).resolve().parent
     systemd_dir = Path('/etc/systemd/system')
 
     # Define the source and target paths
-    socket_file = 'example.socket'
+    name = 'example'
-    service_file = 'example.service'
+    socket_file = f'{name}.socket'
+    service_file = f'{name}.service'
 
     # Create symlinks
     create_symlink(repo_dir / socket_file, systemd_dir / socket_file)
     create_symlink(repo_dir / service_file, systemd_dir / service_file)
 
     # Reload systemd, start and enable the socket
-    run_command('sudo systemctl daemon-reload')
+    try:
+        cmd_kwargs = dict(check=True, text=True, capture_output=True)
+        subprocess.run(['sudo', 'systemctl', 'daemon-reload'], **cmd_kwargs)
         print('Reloaded systemd services')
 
-    run_command('sudo systemctl start example.socket')
+        subprocess.run(['sudo', 'systemctl', 'start', socket_file], **cmd_kwargs)
         print(f'Started [blue]{socket_file}[/]')
 
-    run_command('sudo systemctl enable example.socket')
+        subprocess.run(['sudo', 'systemctl', 'enable', socket_file], **cmd_kwargs)
         print(f'Enabled [blue]{socket_file}[/] to start at boot')
+    except subprocess.CalledProcessError as e:
+        print('Error:', e.stderr)
 
 
 if __name__ == '__main__':

requirements.txt

@@ -0,0 +1 @@
+rich

run_service.py

@@ -1,27 +1,30 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python
 
-import os
 import subprocess
 
-from rich import print
 
-os.chdir('/root')
+def run_tunnel():
-cmd = ['code', 'tunnel']
+    process = subprocess.Popen(
-# cmd = ['echo', '"Hello world"']
+        ['/root/code', 'tunnel'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True
-
-
-print('PATH:\n' + '\n'.join(f' - {p}' for p in os.environ['PATH'].split(':')))
-try:
-    result = subprocess.run(
-        cmd,
-        shell=True,
-        env=os.environ,
-        text=True,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
     )
-    result.check_returncode()
+
-except subprocess.CalledProcessError as e:
+    with process.stdout:
-    print(e)
+        try:
+            for line in iter(process.stdout.readline, ''):
+                print(line, end='')
+        except KeyboardInterrupt:
+            print('KeyboardInterrupt')
         finally:
-    print(result.stdout)
+            process.terminate()
+            try:
+                # Give some time to gracefully shutdown
+                returncode = process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                print('Timeout expired, sending SIGKILL')
+                process.kill()
+                returncode = process.wait()
+    print(f'Process finished with {returncode}')
+
+
+if __name__ == '__main__':
+    run_tunnel()

test_socket.py

@@ -9,16 +9,10 @@ console = Console()
 print = console.print
 
 with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
-    try:
     sock.connect('/run/example.sock')
-    except Exception:
-        raise
-    else:
     with console.status('Connected...'):
         while True:
             try:
                 sleep(0.1)
             except KeyboardInterrupt:
                 break
-    finally:
-        print('Exiting')