drive.

src/snekssh/app.py

@@ -0,0 +1,71 @@
+import asyncio
+import asyncssh
+import os
+import logging
+asyncssh.set_debug_level(2)
+logging.basicConfig(level=logging.DEBUG)
+# Configuration for SFTP server
+SFTP_ROOT = "."  # Directory to serve
+USERNAME = "test"
+PASSWORD = "woeii"
+HOST = "localhost"
+PORT = 2225
+
+class MySFTPServer(asyncssh.SFTPServer):
+    def __init__(self, chan):
+        super().__init__(chan)
+        self.root = os.path.abspath(SFTP_ROOT)
+
+    async def stat(self, path):
+        """Handles 'stat' command from SFTP client"""
+        full_path = os.path.join(self.root, path.lstrip("/"))
+        return await super().stat(full_path)
+
+    async def open(self, path, flags, attrs):
+        """Handles file open requests"""
+        full_path = os.path.join(self.root, path.lstrip("/"))
+        return await super().open(full_path, flags, attrs)
+
+    async def listdir(self, path):
+        """Handles directory listing"""
+        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')}")
+
+    def connection_lost(self, exc):
+        print("Client disconnected")
+
+    def begin_auth(self, username):
+        return True  # No additional authentication steps
+
+    def password_auth_supported(self):
+        return True  # Support password authentication
+
+    def validate_password(self, 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
+
+    await asyncssh.create_server(
+        lambda: MySSHServer(),
+        host=HOST,
+        port=PORT,
+        server_host_keys=["ssh_host_key"],
+        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())
+    except (OSError, asyncssh.Error) as e:
+        print(f"Error starting SFTP server: {e}")