straywild/network/network_chat_protocol.gd

class_name NetworkChatProtocol
extends RefCounted

const CAPABILITY: StringName = &"chat_v1"
const RELIABLE_CHANNEL: int = NetworkProtocol.CHAT_RELIABLE_CHANNEL
const MAX_VISIBLE_CHARACTERS: int = 256
const MAX_UTF8_BYTES: int = 1024
const MAX_HISTORY: int = 100
const LATE_JOIN_HISTORY: int = 50
const MAX_ID_LENGTH: int = 96

static var _invisible_edge_pattern: RegEx = RegEx.create_from_string(
	"^[\\s\\p{Z}\\p{Cf}\\x{2800}]+|[\\s\\p{Z}\\p{Cf}\\x{2800}]+$"
)
static var _visible_content_pattern: RegEx = RegEx.create_from_string(
	"[^\\s\\p{Z}\\p{Cf}\\x{2800}]"
)
static var _roleplay_separator_pattern: RegEx = RegEx.create_from_string(
	"^[\\s\\p{Z}\\p{Cf}\\x{2800}]"
)

enum Kind { PLAYER, SYSTEM }


static func sanitize_body(value: Variant) -> String:
	if typeof(value) != TYPE_STRING:
		return ""
	var result: String = str(value).replace("\r", " ").replace("\n", " ")
	result = result.replace("\t", " ")
	result = _invisible_edge_pattern.sub(result, "", true)
	if (
		result.is_empty()
		or _visible_content_pattern.search(result) == null
		or result.length() > MAX_VISIBLE_CHARACTERS
	):
		return ""
	if result.to_utf8_buffer().size() > MAX_UTF8_BYTES:
		return ""
	for index: int in result.length():
		var codepoint: int = result.unicode_at(index)
		if codepoint < 32 or codepoint == 127:
			return ""
	return result


static func is_roleplay_body(value: Variant) -> bool:
	var body: String = sanitize_body(value)
	if body.length() < 3 or body.left(3).to_lower() != "/me":
		return false
	return (
		body.length() == 3
		or _roleplay_separator_pattern.search(body.substr(3, 1)) != null
	)


static func roleplay_action(value: Variant) -> String:
	var body: String = sanitize_body(value)
	if not is_roleplay_body(body):
		return ""
	return sanitize_body(body.substr(3))


static func valid_user_body(value: Variant) -> bool:
	var body: String = sanitize_body(value)
	return (
		not body.is_empty()
		and (
			not is_roleplay_body(body)
			or not roleplay_action(body).is_empty()
		)
	)


static func validate_request(data: Variant) -> bool:
	return (
		typeof(data) == TYPE_DICTIONARY
		and _valid_id(data.get("request_id"))
		and _valid_id(data.get("session_id"))
		and valid_user_body(data.get("body"))
		and NetworkIdentityCrypto.valid_fingerprint(
			data.get("sender_fingerprint")
		)
		and typeof(data.get("sender_signature")) == TYPE_PACKED_BYTE_ARRAY
		and PackedByteArray(data["sender_signature"]).size()
			<= NetworkIdentityCrypto.MAX_SIGNATURE_BYTES
	)


static func validate_message(data: Variant) -> bool:
	return (
		typeof(data) == TYPE_DICTIONARY
		and _valid_id(data.get("message_id"))
		and _valid_id(data.get("session_id"))
		and typeof(data.get("sequence")) == TYPE_INT
		and int(data["sequence"]) >= 0
		and typeof(data.get("kind")) == TYPE_INT
		and int(data["kind"]) in [Kind.PLAYER, Kind.SYSTEM]
		and typeof(data.get("sender_peer_id")) == TYPE_INT
		and typeof(data.get("sender_display_name")) == TYPE_STRING
		and str(data["sender_display_name"]).length() <= 24
		and (
			(
				int(data["kind"]) == Kind.SYSTEM
				and not sanitize_body(data.get("body")).is_empty()
			)
			or (
				int(data["kind"]) == Kind.PLAYER
				and valid_user_body(data.get("body"))
			)
		)
		and NetworkIdentityCrypto.valid_fingerprint(
			data.get("sender_fingerprint")
		)
		and typeof(data.get("sender_signature")) == TYPE_PACKED_BYTE_ARRAY
	)


static func signature_fields(data: Dictionary) -> Array:
	return [
		str(data.get("session_id", "")),
		str(data.get("request_id", "")),
		str(data.get("sender_fingerprint", "")),
		sanitize_body(data.get("body")),
	]


static func _valid_id(value: Variant) -> bool:
	return (
		typeof(value) == TYPE_STRING
		and not str(value).is_empty()
		and str(value).length() <= MAX_ID_LENGTH
	)