feat: integrate WebDAV application and add lxml dependency for XML parsing

Add WebdavApplication to the app router and include lxml in project dependencies to support WebDAV protocol functionality. Also perform minor code style cleanups across multiple modules (quote normalization, import reordering, formatting adjustments) without behavioral changes.

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()