72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
|
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}")
|