jtftp/examples/main.py

92 lines
2.9 KiB
Python
Raw Normal View History

2022-07-07 12:37:41 -05:00
# JTFTP - Python/AsyncIO TFTP Server
# Copyright (C) 2022 Jeffrey C. Ollie
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import asyncio
import functools
import logging
import pathlib
import sys
import traceback
import arrow
from jtftp.filesystem.inmemory import InMemoryFilesystem
from jtftp.protocol import tftp_server_protocol_factory
logger = logging.getLogger(__name__)
class Formatter(logging.Formatter):
def __init__(self):
logging.Formatter.__init__(self, "")
def format(self, record: logging.LogRecord):
message = record.msg
if len(record.args) > 0:
message = record.msg % record.args
created = arrow.get(record.created).to("America/Chicago")
name = record.name
if len(name) > 25:
name = name[:24] + "\u2026"
msg = f"{created:YYYY-MM-DD HH:mm:ss.SSSSSS Z} [{name:^25s}] {record.levelname:8s} {message:s}"
if record.exc_info is not None:
msg += "\n"
lines = traceback.format_exception(*record.exc_info)
for line in lines:
msg += line
return msg.rstrip()
def formatTime(self, record: logging.LogRecord, datefmt=None): # noqa: N802
"""Format the timestamp."""
created = arrow.get(record.created)
return created.format("YYYY-MM-DD HH:mm:ss.SSSSSS Z")
def setup_logging() -> None:
"""Set up logging."""
formatter = Formatter()
console = logging.StreamHandler(stream=sys.stderr)
console.setFormatter(formatter)
console.setLevel(logging.DEBUG)
root = logging.getLogger()
for handler in root.handlers:
root.removeHandler(handler)
handler.close()
root.addHandler(console)
root.setLevel(logging.DEBUG)
logging.getLogger("opentelemetry").setLevel(logging.DEBUG)
logging.getLogger("urllib3.connectionpool").setLevel(logging.DEBUG)
logging.getLogger("cisco.connection").setLevel(logging.DEBUG)
def serve():
setup_logging()
# filesystem = TestFilesystem()
filesystem = InMemoryFilesystem()
loop = asyncio.get_event_loop()
future = loop.create_datagram_endpoint(
functools.partial(
tftp_server_protocol_factory, filesystem=filesystem, loop=loop
),
local_addr=("::", 69),
)
loop.run_until_complete(future)
loop.run_forever()