BordedDev/snek
1
0
Fork 0
forked from retoor/snek
Code Pull Requests Activity

feat: add initial Dockerfile, SSH server apps, and media upload web component

Browse Source 
Add DockerfileDrive for containerized deployment with Python 3.12.8 Alpine base, including SSH host key and pip setup. Implement multiple SSH server variants: app.py with SFTP file transfer support on port 2225, app2.py with custom bash shell spawning, app3.py with terminal resize handling, app4.py with bcrypt password authentication, and app5.py with chat client infrastructure. Include TileGridElement and UploadButton custom web components in media-upload.js for gallery-style image display with hover effects.
This commit is contained in:
retoor
2025-01-31 11:06:42 +00:00
parent e290a9f5cb
commit 6371bb1fa5
9 changed files with 604 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/snek/static/media-upload.js
+167
View File
@@ -0,0 +1,167 @@
class TileGridElement extends HTMLElement {
constructor() {
super();
this.attachShadow({mode: 'open'});
this.gridId = this.getAttribute('grid');
this.component = document.createElement('div');
this.shadowRoot.appendChild(this.component)
}
connectedCallback() {
console.log('connected');
this.styleElement = document.createElement('style');
this.styleElement.innerText = `
.grid {
padding: 10px;
display: flex;
flex-wrap: wrap;
gap: 10px;
justify-content: center;
}
.grid .tile {
margin: 10px;
width: 100px;
height: 100px;
background-color: #ccc;
object-fit: cover;
border-radius: 10px;
transition: transform 0.3s ease-in-out;
}
.grid .tile:hover {
transform: scale(1.1);
}
`;
this.component.appendChild(this.styleElement);
this.container = document.createElement('div');
this.container.classList.add('gallery');
this.component.appendChild(this.container);
}
addImage(src) {
const item = document.createElement('img');
item.src = src;
item.classList.add('tile');
item.style.width = '100px';
item.style.height = '100px';
this.container.appendChild(item);
}
addImages(srcs) {
srcs.forEach(src => this.addImage(src));
}
addElement(element) {
element.cclassList.add('tile');
this.container.appendChild(element);
}
}
class UploadButton extends HTMLElement {
constructor() {
super();
this.attachShadow({mode: 'open'});
this.component = document.createElement('div');
this.shadowRoot.appendChild(this.component)
window.u = this
}
get gridSelector(){
return this.getAttribute('grid');
}
grid = null
addImages(urls) {
this.grid.addImages(urls);
}
connectedCallback()
{
console.log('connected');
this.styleElement = document.createElement('style');
this.styleElement.innerHTML = `
.upload-button {
display: flex;
flex-direction: column;
align-items: center;
justify-content: center;
gap: 10px;
}
.upload-button input[type="file"] {
display: none;
}
.upload-button label {
display: block;
padding: 10px 20px;
background-color: #ccc;
border-radius: 5px;
cursor: pointer;
}
.upload-button label:hover {
background-color: #999;
}
`;
this.shadowRoot.appendChild(this.styleElement);
this.container = document.createElement('div');
this.container.classList.add('upload-button');
this.shadowRoot.appendChild(this.container);
const input = document.createElement('input');
input.type = 'file';
input.accept = 'image/*';
input.multiple = true;
input.addEventListener('change', (e) => {
const files = e.target.files;
const urls = [];
for (let i = 0; i < files.length; i++) {
const file = files[i];
const reader = new FileReader();
reader.onload = (e) => {
urls.push(e.target.result);
if (urls.length === files.length) {
this.addImages(urls);
}
};
reader.readAsDataURL(file);
}
});
const label = document.createElement('label');
label.textContent = 'Upload Images';
label.appendChild(input);
this.container.appendChild(label);
}
}
customElements.define('upload-button', UploadButton);
customElements.define('tile-grid', TileGridElement);
class MeniaUploadElement extends HTMLElement {
constructor(){
super()
this.attachShadow({mode:'open'})
this.component = document.createElement("div")
alert('aaaa')
this.shadowRoot.appendChild(this.component)
}
connectedCallback() {
this.container = document.createElement("div")
this.component.style.height = '100%'
this.component.style.backgroundColor ='blue';
this.shadowRoot.appendChild(this.container)
this.tileElement = document.createElement("tile-grid")
this.tileElement.style.backgroundColor = 'red'
this.tileElement.style.height = '100%'
this.component.appendChild(this.tileElement)
this.uploadButton = document.createElement('upload-button')
this.component.appendChild(this.uploadButton)
// const mediaUpload = document.createElement('media-upload')
//this.component.appendChild(mediaUpload)
}
}
customElements.define('menia-upload', MeniaUploadElement)
src/snekssh/app.py
+71
View File
@@ -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}")
src/snekssh/app2.py
+68
View File
@@ -0,0 +1,68 @@
import asyncio
import asyncssh
import os
# SSH Server Configuration
HOST = "0.0.0.0"
PORT = 2225
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')}")
def connection_lost(self, exc):
print("Client disconnected")
def password_auth_supported(self):
return True
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()
env["TERM"] = "xterm-256color"
# 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
)
async def read_output():
while True:
data = await bash_proc.stdout.read(1)
if not data:
break
process.stdout.write(data)
async def read_input():
while True:
data = await process.stdin.read(1)
if not data:
break
bash_proc.stdin.write(data)
await asyncio.gather(read_output(), read_input())
async def start_ssh_server():
"""Starts the AsyncSSH server with Bash"""
await asyncssh.create_server(
lambda: CustomSSHServer(),
host=HOST,
port=PORT,
server_host_keys=["ssh_host_key"],
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
+71
View File
@@ -0,0 +1,71 @@
#!/usr/bin/env python3.7
#
# Copyright (c) 2013-2024 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
# To run this program, the file ``ssh_host_key`` must exist with an SSH
# 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``.
#
# 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
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}')
if pixwidth and pixheight:
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}')
if exc.pixwidth and exc.pixheight:
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)
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))
loop.run_forever()
src/snekssh/app4.py
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python3.7
#
# Copyright (c) 2013-2024 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
# To run this program, the file ``ssh_host_key`` must exist with an SSH
# 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
from typing import Optional
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)
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}.')
def connection_lost(self, exc: Optional[Exception]) -> None:
if exc:
print('SSH connection error: ' + str(exc), file=sys.stderr)
else:
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''
def password_auth_supported(self) -> bool:
return True
def validate_password(self, username: str, password: str) -> bool:
if username not in passwords:
return False
pw = passwords[username]
if not password and not pw:
return True
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)
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))
loop.run_forever()
src/snekssh/app5.py
+106
View File
@@ -0,0 +1,106 @@
#!/usr/bin/env python3.7
#
# Copyright (c) 2016-2024 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
# To run this program, the file ``ssh_host_key`` must exist with an SSH
# 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``.
#
# 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
from typing import List, cast
class ChatClient:
_clients: List['ChatClient'] = []
def __init__(self, process: asyncssh.SSHServerProcess):
self._process = process
@classmethod
async def handle_client(cls, process: asyncssh.SSHServerProcess):
await cls(process).run()
async def readline(self) -> str:
return cast(str, self._process.stdin.readline())
def write(self, msg: str) -> None:
self._process.stdout.write(msg)
def broadcast(self, msg: str) -> None:
for client in self._clients:
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''
def password_auth_supported(self) -> bool:
return True
def validate_password(self, username: str, password: str) -> bool:
#if username not in passwords:
# return False
#pw = passwords[username]
#if not password and not pw:
# return True
return True
#return bcrypt.checkpw(password.encode('utf-8'), pw)
async def run(self) -> None:
self.write('Welcome to chat!\n\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._clients.append(self)
self.broadcast(f'*** {name} has entered chat ***\n')
try:
async for line in self._process.stdin:
self.broadcast(f'{name}: {line}')
except asyncssh.BreakReceived:
pass
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)
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))
loop.run_forever()