Author | SHA1 | Message | Date |
---|---|---|---|
Forest Belton | 4fc22a1722 | Implement stubbed MODE handler | 2 years ago |
Forest Belton | 49278f9432 | Fix format of WHO reply | 2 years ago |
Forest Belton | c8805d0a65 | Implement WHO command | 2 years ago |
Forest Belton | 8824a465ce | Update channel indexes when user changes nickname | 2 years ago |
@ -1,2 +1,3 @@ | |||
[flake8] | |||
ignore = E203 | |||
max-line-length = 100 |
@ -0,0 +1,27 @@ | |||
import logging | |||
from paircd.reply import RPL_CHANNELMODEIS | |||
from paircd.client import Client | |||
from paircd.command_handler import CommandHandler | |||
from paircd.message import Message | |||
from paircd.server import Server | |||
class ModeHandler(CommandHandler): | |||
def __init__(self) -> None: | |||
super().__init__("MODE") | |||
async def handle(self, server: Server, client: Client, msg: Message) -> None: | |||
name = msg.args[0] | |||
if not name.startswith("#"): | |||
client.log("TODO: implement user mode queries", level=logging.WARN) | |||
return | |||
channel = server.get_channel_by_name(name) | |||
modes = "".join(channel.get_modes()) | |||
if modes != "": | |||
modes = f"+{modes}" | |||
client.log("TODO: implement channel modes", level=logging.WARN) | |||
msg = RPL_CHANNELMODEIS(client.nickname, name, modes, "") | |||
client.write_message(msg) |
@ -0,0 +1,48 @@ | |||
import logging | |||
from paircd.reply import RPL_ENDOFWHO, RPL_WHOREPLY | |||
from paircd.client import Client | |||
from paircd.command_handler import CommandHandler | |||
from paircd.message import Message | |||
from paircd.server import Server | |||
class WhoHandler(CommandHandler): | |||
def __init__(self) -> None: | |||
super().__init__("WHO", 1) | |||
async def handle(self, server: Server, client: Client, msg: Message) -> None: | |||
name = msg.args[0] | |||
if not name.startswith("#"): | |||
client.log("used WHO on invalid channel", level=logging.WARN) | |||
self.who_end(client, name) | |||
return | |||
if name not in client.channels: | |||
client.log("client used WHO outside of channel", level=logging.WARN) | |||
self.who_end(client, name) | |||
return | |||
channel = server.get_channel_by_name(name) | |||
for member in channel.clients_by_nick.values(): | |||
self.who_reply(client, name, member) | |||
self.who_end(client, name) | |||
def who_reply(self, client: Client, channel_name: str, member: Client) -> None: | |||
msg = RPL_WHOREPLY( | |||
client.nickname, | |||
channel_name, | |||
member.username, | |||
member.hostname, | |||
"0", # server | |||
member.nickname, | |||
# TODO: Implement | |||
"H", # H = here, G = gone | |||
"0", # hop count | |||
member.realname, | |||
) | |||
client.write_message(msg) | |||
def who_end(self, client: Client, channel_name: str) -> None: | |||
msg = RPL_ENDOFWHO(client.nickname, channel_name) | |||
client.write_message(msg) |
@ -1,7 +0,0 @@ | |||
import logging | |||
from paircd.client import Client | |||
def log_client(client: Client, msg: str, level: int = logging.INFO) -> None: | |||
logging.log(level, f"{client.hostname} ({client.id()}) {msg}") |
@ -0,0 +1,23 @@ | |||
from typing import Any, Callable, List | |||
from paircd.message import Message | |||
def reply_fn(cmd: int, tmpl: str) -> Callable: | |||
def fn(target: str, *args: List[Any]) -> Message: | |||
msg = f"{target} {tmpl.format(*args)}" | |||
return Message(cmd=str(cmd), args=[msg]) | |||
return fn | |||
# Error replies | |||
ERR_NOSUCHNICK = reply_fn(401, "{0} :No such nick/channel") | |||
ERR_NOSUCHSERVER = reply_fn(402, "{0}: No such server") | |||
# Command responses | |||
RPL_WHOISUSER = reply_fn(311, "{0} {1} {2} * :{3}") | |||
RPL_ENDOFWHO = reply_fn(315, "{0} :End of /WHO list") | |||
RPL_ENDOFWHOIS = reply_fn(318, "{0} :End of /WHOIS list") | |||
RPL_CHANNELMODEIS = reply_fn(324, "{0} {1}{2}") | |||
RPL_WHOREPLY = reply_fn(352, "{0} {1} {2} {3} {4} {5} :{6} {7}") |