snek/src/snekssh/app.py

_A=True
import asyncio,logging,os,asyncssh
asyncssh.set_debug_level(2)
logging.basicConfig(level=logging.DEBUG)
SFTP_ROOT='.'
USERNAME='test'
PASSWORD='woeii'
HOST='localhost'
PORT=2225
class MySFTPServer(asyncssh.SFTPServer):
	def __init__(A,chan):super().__init__(chan);A.root=os.path.abspath(SFTP_ROOT)
	async def stat(A,path):"Handles 'stat' command from SFTP client";B=os.path.join(A.root,path.lstrip('/'));return await super().stat(B)
	async def open(A,path,flags,attrs):'Handles file open requests';B=os.path.join(A.root,path.lstrip('/'));return await super().open(B,flags,attrs)
	async def listdir(A,path):'Handles directory listing';B=os.path.join(A.root,path.lstrip('/'));return await super().listdir(B)
class MySSHServer(asyncssh.SSHServer):
	'Custom SSH server to handle authentication'
	def connection_made(A,conn):print(f"New connection from {conn.get_extra_info("peername")}")
	def connection_lost(A,exc):print('Client disconnected')
	def begin_auth(A,username):return _A
	def password_auth_supported(A):return _A
	def validate_password(C,username,password):A=password;B=username;print(B,A);return _A;return B==USERNAME and A==PASSWORD
async def start_sftp_server():os.makedirs(SFTP_ROOT,exist_ok=_A);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()
if __name__=='__main__':
	try:asyncio.run(start_sftp_server())
	except(OSError,asyncssh.Error)as e:print(f"Error starting SFTP server: {e}")