Added webdav.

src/snekssh/app.py

@@ -1,7 +1,9 @@
 import asyncio
-import asyncssh
-import os
 import logging
+import os
+
+import asyncssh
+
 asyncssh.set_debug_level(2)
 logging.basicConfig(level=logging.DEBUG)
 # Configuration for SFTP server
@@ -11,6 +13,7 @@ PASSWORD = "woeii"
 HOST = "localhost"
 PORT = 2225
 
+
 class MySFTPServer(asyncssh.SFTPServer):
     def __init__(self, chan):
         super().__init__(chan)
@@ -31,8 +34,10 @@ class MySFTPServer(asyncssh.SFTPServer):
         full_path = os.path.join(self.root, path.lstrip("/"))
         return await super().listdir(full_path)
 
+
 class MySSHServer(asyncssh.SSHServer):
     """Custom SSH server to handle authentication"""
+
     def connection_made(self, conn):
         print(f"New connection from {conn.get_extra_info('peername')}")
 
@@ -46,11 +51,12 @@ class MySSHServer(asyncssh.SSHServer):
         return True  # Support password authentication
 
     def validate_password(self, username, password):
-        print(username,password)
-        
+        print(username, password)
+
         return True
         return username == USERNAME and password == PASSWORD
 
+
 async def start_sftp_server():
     os.makedirs(SFTP_ROOT, exist_ok=True)  # Ensure the root directory exists
 
@@ -59,11 +65,12 @@ async def start_sftp_server():
         host=HOST,
         port=PORT,
         server_host_keys=["ssh_host_key"],
-        process_factory=MySFTPServer
+        process_factory=MySFTPServer,
     )
     print(f"SFTP server running on {HOST}:{PORT}")
     await asyncio.Future()  # Keep running forever
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(start_sftp_server())

src/snekssh/app2.py

@@ -1,7 +1,8 @@
 import asyncio
-import asyncssh
 import os
 
+import asyncssh
+
 # SSH Server Configuration
 HOST = "0.0.0.0"
 PORT = 2225
@@ -9,6 +10,7 @@ USERNAME = "user"
 PASSWORD = "password"
 SHELL = "/bin/sh"  # Change to another shell if needed
 
+
 class CustomSSHServer(asyncssh.SSHServer):
     def connection_made(self, conn):
         print(f"New connection from {conn.get_extra_info('peername')}")
@@ -22,6 +24,7 @@ class CustomSSHServer(asyncssh.SSHServer):
     def validate_password(self, username, password):
         return username == USERNAME and password == PASSWORD
 
+
 async def custom_bash_process(process):
     """Spawns a custom bash shell process"""
     env = os.environ.copy()
@@ -29,7 +32,12 @@ async def custom_bash_process(process):
 
     # Start the Bash shell
     bash_proc = await asyncio.create_subprocess_exec(
-        SHELL, "-i", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env
+        SHELL,
+        "-i",
+        stdin=asyncio.subprocess.PIPE,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        env=env,
     )
 
     async def read_output():
@@ -48,6 +56,7 @@ async def custom_bash_process(process):
 
     await asyncio.gather(read_output(), read_input())
 
+
 async def start_ssh_server():
     """Starts the AsyncSSH server with Bash"""
     await asyncssh.create_server(
@@ -55,14 +64,14 @@ async def start_ssh_server():
         host=HOST,
         port=PORT,
         server_host_keys=["ssh_host_key"],
-        process_factory=custom_bash_process
+        process_factory=custom_bash_process,
     )
     print(f"SSH server running on {HOST}:{PORT}")
     await asyncio.Future()  # Keep running
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(start_ssh_server())
     except (OSError, asyncssh.Error) as e:
         print(f"Error starting SSH server: {e}")
-

src/snekssh/app3.py

@@ -27,45 +27,48 @@
 # The file ``ssh_user_ca`` must exist with a cert-authority entry of
 # the certificate authority which can sign valid client certificates.
 
-import asyncio, asyncssh, sys
+import asyncio
+import sys
+
+import asyncssh
+
 
 async def handle_client(process: asyncssh.SSHServerProcess) -> None:
 
-
-
-
-
     width, height, pixwidth, pixheight = process.term_size
 
-    process.stdout.write(f'Terminal type: {process.term_type}, '
-                         f'size: {width}x{height}')
+    process.stdout.write(
+        f"Terminal type: {process.term_type}, " f"size: {width}x{height}"
+    )
     if pixwidth and pixheight:
-        process.stdout.write(f' ({pixwidth}x{pixheight} pixels)')
-    process.stdout.write('\nTry resizing your window!\n')
+        process.stdout.write(f" ({pixwidth}x{pixheight} pixels)")
+    process.stdout.write("\nTry resizing your window!\n")
 
     while not process.stdin.at_eof():
         try:
             await process.stdin.read()
         except asyncssh.TerminalSizeChanged as exc:
-            process.stdout.write(f'New window size: {exc.width}x{exc.height}')
+            process.stdout.write(f"New window size: {exc.width}x{exc.height}")
             if exc.pixwidth and exc.pixheight:
-                process.stdout.write(f' ({exc.pixwidth}'
-                                     f'x{exc.pixheight} pixels)')
-            process.stdout.write('\n')
-
-
+                process.stdout.write(f" ({exc.pixwidth}" f"x{exc.pixheight} pixels)")
+            process.stdout.write("\n")
 
 
 async def start_server() -> None:
-    await asyncssh.listen('', 2230, server_host_keys=['ssh_host_key'],
-                          #authorized_client_keys='ssh_user_ca',
-                          process_factory=handle_client)
+    await asyncssh.listen(
+        "",
+        2230,
+        server_host_keys=["ssh_host_key"],
+        # authorized_client_keys='ssh_user_ca',
+        process_factory=handle_client,
+    )
+
 
 loop = asyncio.new_event_loop()
 
 try:
     loop.run_until_complete(start_server())
 except (OSError, asyncssh.Error) as exc:
-    sys.exit('Error starting server: ' + str(exc))
+    sys.exit("Error starting server: " + str(exc))
 
 loop.run_forever()

src/snekssh/app4.py

@@ -24,32 +24,39 @@
 # private key in it to use as a server host key. An SSH host certificate
 # can optionally be provided in the file ``ssh_host_key-cert.pub``.
 
-import asyncio, asyncssh, bcrypt, sys
+import asyncio
+import sys
 from typing import Optional
 
-passwords = {'guest': b'',                # guest account with no password
-             'user': bcrypt.hashpw(b'user', bcrypt.gensalt()),
-            }
+import asyncssh
+import bcrypt
+
+passwords = {
+    "guest": b"",  # guest account with no password
+    "user": bcrypt.hashpw(b"user", bcrypt.gensalt()),
+}
+
 
 def handle_client(process: asyncssh.SSHServerProcess) -> None:
-    username = process.get_extra_info('username')
-    process.stdout.write(f'Welcome to my SSH server, {username}!\n')
-    #process.exit(0)
+    username = process.get_extra_info("username")
+    process.stdout.write(f"Welcome to my SSH server, {username}!\n")
+    # process.exit(0)
+
 
 class MySSHServer(asyncssh.SSHServer):
     def connection_made(self, conn: asyncssh.SSHServerConnection) -> None:
-        peername = conn.get_extra_info('peername')[0]
-        print(f'SSH connection received from {peername}.')
+        peername = conn.get_extra_info("peername")[0]
+        print(f"SSH connection received from {peername}.")
 
     def connection_lost(self, exc: Optional[Exception]) -> None:
         if exc:
-            print('SSH connection error: ' + str(exc), file=sys.stderr)
+            print("SSH connection error: " + str(exc), file=sys.stderr)
         else:
-            print('SSH connection closed.')
+            print("SSH connection closed.")
 
     def begin_auth(self, username: str) -> bool:
         # If the user's password is the empty string, no auth is required
-        return passwords.get(username) != b''
+        return passwords.get(username) != b""
 
     def password_auth_supported(self) -> bool:
         return True
@@ -60,18 +67,24 @@ class MySSHServer(asyncssh.SSHServer):
         pw = passwords[username]
         if not password and not pw:
             return True
-        return bcrypt.checkpw(password.encode('utf-8'), pw)
+        return bcrypt.checkpw(password.encode("utf-8"), pw)
+
 
 async def start_server() -> None:
-    await asyncssh.create_server(MySSHServer, '', 2231,
-                                 server_host_keys=['ssh_host_key'],
-                                 process_factory=handle_client)
+    await asyncssh.create_server(
+        MySSHServer,
+        "",
+        2231,
+        server_host_keys=["ssh_host_key"],
+        process_factory=handle_client,
+    )
+
 
 loop = asyncio.new_event_loop()
 
 try:
     loop.run_until_complete(start_server())
 except (OSError, asyncssh.Error) as exc:
-    sys.exit('Error starting server: ' + str(exc))
+    sys.exit("Error starting server: " + str(exc))
 
 loop.run_forever()

src/snekssh/app5.py

@@ -27,11 +27,15 @@
 # The file ``ssh_user_ca`` must exist with a cert-authority entry of
 # the certificate authority which can sign valid client certificates.
 
-import asyncio, asyncssh, sys
+import asyncio
+import sys
 from typing import List, cast
 
+import asyncssh
+
+
 class ChatClient:
-    _clients: List['ChatClient'] = []
+    _clients: List["ChatClient"] = []
 
     def __init__(self, process: asyncssh.SSHServerProcess):
         self._process = process
@@ -40,8 +44,6 @@ class ChatClient:
     async def handle_client(cls, process: asyncssh.SSHServerProcess):
         await cls(process).run()
 
-    
-
     async def readline(self) -> str:
         return cast(str, self._process.stdin.readline())
 
@@ -53,54 +55,58 @@ class ChatClient:
             if client != self:
                 client.write(msg)
 
-
     def begin_auth(self, username: str) -> bool:
-            # If the user's password is the empty string, no auth is required
-            #return False
-            return True # passwords.get(username) != b''
+        # If the user's password is the empty string, no auth is required
+        # return False
+        return True  # passwords.get(username) != b''
 
     def password_auth_supported(self) -> bool:
         return True
 
     def validate_password(self, username: str, password: str) -> bool:
-        #if username not in passwords:
+        # if username not in passwords:
         #    return False
-        #pw = passwords[username]
-        #if not password and not pw:
+        # pw = passwords[username]
+        # if not password and not pw:
         #    return True
         return True
-        #return bcrypt.checkpw(password.encode('utf-8'), pw)
-
+        # return bcrypt.checkpw(password.encode('utf-8'), pw)
 
     async def run(self) -> None:
-        self.write('Welcome to chat!\n\n')
+        self.write("Welcome to chat!\n\n")
 
-        self.write('Enter your name: ')
-        name = (await self.readline()).rstrip('\n')
+        self.write("Enter your name: ")
+        name = (await self.readline()).rstrip("\n")
 
-        self.write(f'\n{len(self._clients)} other users are connected.\n\n')
+        self.write(f"\n{len(self._clients)} other users are connected.\n\n")
 
         self._clients.append(self)
-        self.broadcast(f'*** {name} has entered chat ***\n')
+        self.broadcast(f"*** {name} has entered chat ***\n")
 
         try:
             async for line in self._process.stdin:
-                self.broadcast(f'{name}: {line}')
+                self.broadcast(f"{name}: {line}")
         except asyncssh.BreakReceived:
             pass
 
-        self.broadcast(f'*** {name} has left chat ***\n')
+        self.broadcast(f"*** {name} has left chat ***\n")
         self._clients.remove(self)
 
+
 async def start_server() -> None:
-    await asyncssh.listen('', 2235, server_host_keys=['ssh_host_key'],
-                          process_factory=ChatClient.handle_client)
+    await asyncssh.listen(
+        "",
+        2235,
+        server_host_keys=["ssh_host_key"],
+        process_factory=ChatClient.handle_client,
+    )
+
 
 loop = asyncio.new_event_loop()
 
 try:
     loop.run_until_complete(start_server())
 except (OSError, asyncssh.Error) as exc:
-    sys.exit('Error starting server: ' + str(exc))
+    sys.exit("Error starting server: " + str(exc))
 
 loop.run_forever()