crystal/tcpsocket.cr: initial work on semi-public socket protocol

1 files changed, 528 insertions, 0 deletions

diff --git a/crystal/tcpsocket.cr b/crystal/tcpsocket.cr
new file mode 100644
index 0000000..8d26a68
--- /dev/null
+++ b/crystal/tcpsocket.cr
@@ -0,0 +1,528 @@
+require "socket"
+require "openssl"
+require "twitch/irc"
+require "http"
+require "uri"
+require "twitcr"
+require "json"
+require "crystal_mpd"
+require "obswebsocket"
+require "yaml"
+
+STDOUT.sync = true
+STDOUT.flush_on_newline = true
+
+struct Nil
+  def as_s?
+    self
+  end
+  def []?( v : String | Int )
+    self
+  end
+end
+
+regexservice = /(twitch|gamesurge)/
+regexuser    = /[0-9a-zA-Z_]+/
+regexb64     = /[0-9a-fA-F]+/
+regexvoice   = /[0-9a-zA-Z-]+/
+
+module BungmoBott
+  class Config
+    # YAML:
+    #---
+    #statedir: /home/jrayhawk/.local/state/bungmobott/
+    #tempdir: /tmp/bungmobott/
+    #listen: 0.0.0.0:5555
+    #join_channels:
+    #  twitch:
+    #    - bungmonkey
+    #    - dopefish
+    #    # - yackamov
+    #  gamesurge:
+    #    - bungmonkey
+    #twitch_channel_id: 59895482 # saves a lookup
+    #chat_user:
+    #  twitch: bungmonkey
+    #  gamesurge: bungmoBott
+
+    getter configdir :           String
+    getter statedir  :           String
+    getter tempdir   :           String
+    getter listen :              String | Nil
+    getter join_channels : Hash( String, Array( String ) ) | Nil
+    getter twitch_channel_id :            UInt64 | Nil
+    getter twitch_client_id :    String | Nil
+    getter twitch_access_token : String | Nil
+    getter gcloud_token :        String | Nil
+    getter chat_user :     Hash( String, String ) | Nil
+    getter metaconfig :    Hash( String, Hash( String, String | UInt64 | Nil | Hash( String, String | Array( String ) ) ) )
+
+    def initialize( )
+
+      @metaconfig = Hash( String, Hash( String, String | UInt64 | Nil | Hash( String, String | Array( String ) ) ) ).new
+      @metaconfig["default"] =    Hash( String, String | UInt64 | Nil | Hash( String, String | Array( String ) ) ).new
+      @metaconfig["default"]["home"] = Path.home.to_s
+      @configdir = Path.home./("/.config/bungmobott/").to_s
+      @metaconfig["default"]["statedir"] = Path.home./("/.local/state/bungmobott/").to_s
+
+      if ENV["LOCALAPPDATA"]?
+        @configdir = Path.windows( ENV["LOCALAPPDATA"] + "\\bungmobott\\" ).to_s
+        @metaconfig["default"]["statedir"] = Path.windows( ENV["LOCALAPPDATA"] + "\\bungmobott\\state\\" ).to_s
+      end
+
+      @metaconfig["default"]["tempdir"] = "/tmp/bungmobott/"
+
+      ENV["TEMP"]? && ( @metaconfig["default"]["tempdir"] = "#{ENV["TEMP"]}\\bungmobott\\" )
+
+      ENV["XDG_CONFIG_HOME"]? && ( @configdir = ENV["XDG_CONFIG_HOME"] + "/bungmobott/" )
+      #ENV["XDG_DATA_HOME"]? && ( @metaconfig["default"]["datadir"] = ENV["XDG_DATA_HOME"] ) # unused?
+      ENV["XDG_STATE_HOME"]? && ( @metaconfig["default"]["statedir"] = ENV["XDG_STATE_HOME"] + "/bungmobott/" )
+
+      populate( "config"  )
+      populate( "secret" )
+      @metaconfig["default"].merge!(
+        {
+          "chat_user"           => nil,
+          "twitch_channel_id"   => nil,
+          "join_channels"       => nil,
+          "listen"              => nil,
+          "twitch_client_id"    => nil,
+          "twitch_access_token" => nil,
+          "gcloud_token"        => nil,
+        }
+      )
+
+      # FIXME: Maybe add more explicit validation on these...?
+      # FIXME: Might need to do explicit maps on these to get them to typecheck
+      @chat_user           = @metaconfig["config"]["chat_user"]?.as( Hash( String, String ) | Nil )
+      @twitch_channel_id   = @metaconfig["config"]["twitch_channel_id"]?.as( UInt64 )
+      @join_channels       = @metaconfig["config"]["join_channels"]?.as( Hash( String, Array( String ) ) | Nil )
+      @twitch_client_id    = @metaconfig["secret"]["twitch_client_id"]?.as( String )
+      @twitch_access_token = @metaconfig["secret"]["twitch_access_token"]?.as( String )
+      @gcloud_token        = @metaconfig["secret"]["gcloud_token"]?.as( String )
+
+      @listen = ( @metaconfig["config"]["listen"]?.as( String | Nil ) )
+      @statedir = (
+        if @metaconfig["config"]["statedir"]?
+          @metaconfig["config"]["statedir"].as( String )
+        else
+          @metaconfig["default"]["statedir"].as( String )
+        end
+      )
+      @tempdir = (
+        if @metaconfig["config"]["tempdir"]?
+          @metaconfig["config"]["tempdir"].as( String )
+        else
+          @metaconfig["default"]["tempdir"].as( String )
+        end
+      )
+
+      Dir.mkdir_p( @statedir )
+      Dir.mkdir_p( @tempdir  )
+
+    end
+    def populate( configname : String )
+      begin
+        @metaconfig[configname] = Hash( String, String | UInt64 | Nil | Hash( String, String | Array( String ) ) ).from_yaml( File.read( configdir + configname + ".txt" ) )
+      rescue exyaml : YAML::ParseException
+        puts "Failed to parse #{configname}.txt: #{exyaml.message}"
+        exit 4
+      rescue exio : IO::Error
+        puts "#{configname}.txt missing, writing new #{configname}.txt"
+        if configname == "secrets"
+          write( configname, @metaconfig["default"].select{ |k, v| k =~ /twitch_(client_id|access_token)|gcloud_token/ }.to_yaml )
+        elsif configname == "config"
+          write( configname, @metaconfig["default"].reject{ |k, v| k =~ /configdir|home|twitch_(client_id|access_token)|gcloud_token/ }.to_yaml )
+        end
+        # verify we can read it back
+        @metaconfig[configname] = Hash( String, String | UInt64 | Nil | Hash( String, String | Array( String ) ) ).from_yaml( File.read( configdir + configname + ".txt" ) )
+      end
+    end
+    def write( configname : String, content : String )
+      File.write( configdir + "#{configname}.txt", content )
+      Dir.mkdir_p( @configdir )
+    rescue exio : IO::Error
+      puts "ERROR: Unable to write #{configname}.txt: #{exio.message}"
+      exit 7
+    end
+  end
+end
+
+regextwitchuser = /^[0-9a-zA-Z_]+$/
+
+config = BungmoBott::Config.new
+
+# enable direct twitch api?
+
+if ( config.twitch_access_token && config.twitch_client_id )
+  twitchapi = true
+else
+  twitchapi = false
+  config.twitch_access_token || STDERR.puts "Warning: #{config.configdir}secret.txt 'twitch_access_token' is missing; direct Twitch API access disabled."
+  config.twitch_client_id    || STDERR.puts "Warning: #{config.configdir}secret.txt 'twitch_client_id' is missing; direct Twitch API access disabled."
+  unless chat_user = ( config.chat_user && config.chat_user.not_nil!["twitch"]? )
+    if chat_user = ( config.join_channels && config.join_channels.not_nil!["twitch"]?[0]? )
+      STDERR.puts "Warning: #{config.configdir}config.txt 'chat_user: {twitch}' value is missing; using first configured 'join_channels: {twitch}' array value instead: #{chat_user}"
+      if ( config.chat_user )
+        config.chat_user.not_nil!["twitch"] = chat_user
+      end
+    else
+      STDERR.puts "ERROR: 'chat_user: {twitch}' string value and 'join_channels: {twitch}' array value missing."
+      exit 3
+    end
+  end
+end
+
+# enable direct gcloud api?
+if config.gcloud_token
+  gcloud = true
+else
+  gcloud = false
+  STDERR.puts "Warning: #{config.configdir}secret.txt gcloud_token is missing; direct GCS voices disabled."
+end
+
+# enable aws?
+if ! File.exists?( Path.home./("/.aws/credentials") )
+  STDERR.puts "Warning: #{Path.home}/.aws/credentials is missing; AWS voices disabled."
+  aws = false
+elsif ! Process.find_executable( "aws.exe" )
+  STDERR.puts "Warning: aws.exe is missing; AWS voices disabled."
+  aws = false
+else
+  aws = true
+end
+
+# enable effects?
+#effects = Hash( String, String ).new
+#if File.exists?( config.configdir + "/effects.txt" )
+#  File.each_line( config.configdir + "/effects.txt" ) do |line|
+#    effects[ line.downcase ] = line
+#  end
+#end
+
+# FIXME: might need twitch_channel_id eventually ...?
+#if twitchapi
+#  twitchclient = Twitcr::Client.new( defaultsettings )
+#  # derive twitch_channel_id from channel or vice versa
+#  [ "channel", "twitch_channel_id" ].each do |key|
+#    File.exists?( config.configdir + key ) && ( settings[key] = File.read( settings["config.configdir"] + key ).chomp )
+#  end
+#  if      ( config.channel? =~ regextwitchuser ) &&   ( settings["twitch_channel_id"]? =~ /^[0-9]+$/ )
+#  elsif ! ( config.channel? =~ regextwitchuser ) &&   ( settings["twitch_channel_id"]? =~ /^[0-9]+$/ )
+#    config.channel    = twitchclient.user( settings["twitch_channel_id"].to_u64 ).login
+#  elsif   ( config.channel? =~ regextwitchuser ) && ! ( settings["twitch_channel_id"]? =~ /^[0-9]+$/ )
+#    config.twitch_channel_id = twitchclient.user( settings["channel"] ).id.to_s
+#  else
+#    STDERR.puts "ERROR: Missing #{config.configdir}channel and twitch_channel_id configuration keys."
+#    error = true
+#    # exit 2
+#  end
+#  if error == true
+#    {% if flag?(:windows) %}
+#      STDERR.puts "press enter to end program"
+#      gets
+#    {% end %}
+#    exit 1
+#  end
+#end
+
+def generatevoicelistaws( )
+  voices = Array(String).new
+  JSON.parse( `aws polly describe-voices` )["Voices"].as_a.each do | v |
+    voices.push( v["Id"].as_s )
+  end
+  return voices
+end
+
+def generatevoicelistgcs( gcloud_token )
+  voices = Array(String).new
+  ssl_context = OpenSSL::SSL::Context::Client.new
+  headers = HTTP::Headers.new
+  headers["Content-Type"] = "application/json; charset=utf-8"
+  response = HTTP::Client.exec( "GET", "https://texttospeech.googleapis.com/v1/voices?key=#{gcloud_token}", headers, nil, tls: ssl_context )
+  JSON.parse( response.body )["voices"].as_a.each do | v |
+    voices.push( v["name"].as_s )
+  end
+  return voices
+end
+
+{% if flag?(:windows) %}
+def generatevoicelistwin()
+  voices = Array(String).new
+  p = Process.new(
+    "powershell.exe",
+    [ "-Command", "
+      Add-Type -AssemblyName System.Speech;
+      $speak = New-Object System.Speech.Synthesis.SpeechSynthesizer;
+      $speak.GetInstalledVoices().VoiceInfo | Select-Object Name
+    "],
+    output: Process::Redirect::Pipe
+  )
+  p.output.each_line do | v |
+    v = v.gsub(/ +$/, "")
+    v = v.gsub(/ Desktop$/, "")
+    v = v.gsub( " ", "-" )
+    if v =~ /[A-Za-z0-9]-[A-Za-z0-9]/
+      voices.push( v )
+    end
+  end
+  return voices
+end
+{% end %}
+
+def regeneratevoicelist( defaultsettings : Hash( String, String ), aws : Bool, gcloud : Bool )
+  voices = Hash(String, String).new
+  if aws
+    generatevoicelistaws() do | voice |
+      voices[ voice.downcase ] = voice
+    end
+  end
+  if gcloud
+    generatevoicelistgcs( config.gcloud_token ).each do | voice |
+      voices[ voice.downcase ] = voice
+    end
+  end
+  {% if flag?(:windows) %}
+    generatevoicelistwin().each do
+  {% end %}
+
+  File.write( config.configdir + "/voicelist.txt", voices.values.sort.join("\r\n") )
+  return voices
+end
+
+# channel message subscriptions: { service, chan } => [ client, client ]
+channelsubs = Hash( Tuple( String, String ), Array( OpenSSL::SSL::Socket::Server ) ).new
+# Unencrypted
+#connections = Hash(TCPSocket, Hash(String, String)).new
+# Encrypted
+connections = Hash(OpenSSL::SSL::Socket::Server, Hash(String, String)).new
+
+ircipc = Channel( Tuple( String, String ) ).new
+t2sipc = Channel( Tuple( String, String ) ).new
+twitchipc = Channel( Tuple( String, Union( String, UInt64 ) ) ).new
+
+# Twitch API request handling thread
+#spawn do
+#  loop do
+#    begin
+#      while twitchtuple = twitchipc.receive
+#        cmd, arg = [ *twitchtuple ]
+#        case cmd
+#        when "get_user"
+#          userinfo = JSON.parse( twitchclient.get_user( arg ) )["data"][0]
+#          pp userinfo
+#          unless userinfo["broadcaster_type"].as_s.blank?
+#            puts "\033[38;5;12m#{userinfo["login"]} is #{userinfo["broadcaster_type"]}\033[0m"
+#          end
+#          userage = ( Time.utc.to_unix - Time::Format::RFC_3339.parse( userinfo["created_at"].as_s ).to_unix )
+#          if ( userage - 172800 ) < 0
+#            puts "\033[38;5;1m#{userinfo["login"]}'s account is #{((172800 - userage)/60/60).to_i64} hours old.\033[0m"
+#          end
+#        when "get_followers"
+#          followers = JSON.parse( twitchclient.get_user_follows( to: arg.to_u64 ) )["total"].as_i64
+#          if followers > 500
+#            puts "\033[38;5;2m#{followers} followers\033[0m"
+#          end
+#        end
+#      end
+#    end
+#  end
+#end
+#
+
+if ( config.twitch_access_token && config.chat_user.not_nil!["twitch"]? && config.join_channels.not_nil!["twitch"]? )
+  # IRC thread
+  spawn do
+    loop do
+      begin
+        bot = Twitch::IRC::Client.new( nick: config.chat_user.not_nil!["twitch"], token: "oauth:" + config.twitch_access_token.not_nil!, log_mode: true )
+        bot.tags = [ "membership", "tags", "commands" ]
+
+        # Outgoing IRC message thread
+        # "Most IRC servers limit messages to 512 bytes in length, including the trailing CR-LF characters."
+        # PRIVMSG #channel message\r\n
+        spawn do
+          while tuple = ircipc.receive # why does this need to be a tuple?
+            sizelimit=( 512 - ( tuple[0].size + 12 ) )
+            if    ( tuple[0] == "JOIN" )
+              if ( tuple[0] =~ regextwitchuser )
+                bot.join_channel( tuple[1] )
+              else
+                STDERR.puts "Invalid channel name #{ tuple[1] }"
+              end
+            elsif ( tuple[0] =~ /^#/ )
+              bot.message( tuple[0], tuple[1][0..sizelimit] ) # limit size
+            end
+          end
+        end
+
+        # Create a handler to process incoming messages
+        bot.on_message do |message|
+          spawn do
+  #Message(
+  #  @tags={
+  #    "badge-info" => "",
+  #    "badges" => "moderator/1,bits/100",
+  #    "color" => "",
+  #    "display-name" => "BungMonkey",
+  #    "emotes" => "",
+  #    "first-msg" => "0",
+  #    "flags" => "",
+  #    "id" => "3170330a-66dd-4163-a5cb-5a380abef366",
+  #    "mod" => "1",
+  #    "returning-chatter" => "0",
+  #    "room-id" => "22579666",
+  #    "subscriber" => "0",
+  #    "tmi-sent-ts" => "1692002718705",
+  #    "turbo" => "0",
+  #    "user-id" => "59895482",
+  #    "user-type" => "mod"
+  #  },
+  #  @prefix=Prefix(
+  #    @source="bungmonkey",
+  #    @user="bungmonkey",
+  #    @host="bungmonkey.tmi.twitch.tv"
+  #  ),
+  #  @command="PRIVMSG",
+  #  @params=[
+  #    "#tenichi",
+  #    "test"
+  #  ]
+  #)
+
+            channelsubs[{ "twitch", message.params[0] }].each do | client |
+              if ( ( prefix = message.prefix ) && ( chatuser = prefix.source ) ) # && ( uid = message.tags["user-id"]? )
+                client.puts "msg twitch #{message.params[0]} #{chatuser} #{message.params[1]}"
+              end
+            end
+            pp message
+            pp message.params
+            own = ( message.tags["room-id"] == message.tags["user-id"] )
+            vip = ( message.tags["badges"].to_s.matches?( /(^|,)vip\// ) )
+            mod = ( message.tags["mod"] == "1" )
+            sub = ( message.tags["subscriber"] == "1" )
+          rescue ex
+            puts ex
+            #ircipc.send( { "##{config.channel}", "An error occurred! " + ex.message.to_s } )
+            # Maybe send all error messages out through the API? Have to do channel->client mappings, though.
+          end
+        end
+
+        rooms = Array( String ).new
+        rooms = config.join_channels.not_nil!["twitch"]
+
+        # Connect to Twitch
+        bot.run( rooms.map{ | room | room.sub( /^#/, "") } )
+      rescue ex : IO::Error
+        pp ex
+        sleep 1
+        # loop to reconnect
+      rescue ex
+        pp ex
+        {% if flag?(:windows) %}
+          puts "press enter to end program"
+          gets
+        {% end %}
+        exit 1
+      end
+    end
+  end
+end
+
+if config.listen
+  begin
+    ip, port = config.listen.not_nil!.split(":")
+    tcp_server = TCPServer.new( ip, port.to_i )
+    ssl_context = OpenSSL::SSL::Context::Server.new
+    ssl_context.private_key=(config.configdir + "/privkey.pem" )
+    ssl_context.certificate_chain=(config.configdir + "/fullchain.pem" )
+  # unauthenticated config; seems to be broken in crystal v1.9.2
+  #  ssl_context = OpenSSL::SSL::Context::Server.insecure()
+  #  ssl_context.add_options(OpenSSL::SSL::Options::ALL)
+  #  ssl_context.security_level=0
+  #  ssl_context.ciphers=("ADH@SECLEVEL=0")
+  #  ssl_context.ciphers=("ADH-AES256-GCM-SHA384:@SECLEVEL=0")
+
+  # bungmobott protocol server
+    #while tcp_server.accept? do | clientsocket |
+    while clientsocket = tcp_server.accept?
+      spawn do
+        client = OpenSSL::SSL::Socket::Server.new(clientsocket, ssl_context)
+        client.flush_on_newline=true
+
+        puts "Connected: #{clientsocket.remote_address}"
+        connections[client] = Hash(String, String).new
+        connections[client]["remote_address"] = clientsocket.remote_address.to_s
+        connections[client]["authed"] = "false"
+
+        while message = client.gets
+          puts message
+          client.puts message
+          if ( connections[client]["authed"] == "true" )
+            # irc
+            if ( match = message.match( /^irc (#{regexservice}) (#{regexuser})$/i ) )
+              ircservice = match[1]
+              ircchannel = match[3]
+              client.puts "joining #{ircservice} #{ircchannel}"
+              ircipc.send( { "JOIN", ircchannel } )
+              unless channelsubs[ { ircservice, ircchannel } ]?
+                channelsubs[ { ircservice, "#" + ircchannel } ] = Array( OpenSSL::SSL::Socket::Server ).new
+              end
+              channelsubs[ { ircservice, "#" + ircchannel } ].push( client )
+              pp channelsubs
+            elsif ( match = message.match( /^awst2s (#{regexvoice}) (.+)/i ) )
+              # FIXME # powershell doesn't take bytestreams; maybe use http file hosting for this?
+            elsif ( match = message.match( /^gcst2s (#{regexvoice}) (.+)/i ) )
+              # FIXME
+            elsif ( match = message.match( /^awsvoicelist$/i ) )
+              client.puts "awsvoicelist " + generatevoicelistaws().join(" ")
+            elsif ( match = message.match( /^gcsvoicelist$/i ) )
+              client.puts "gcsvoicelist " + generatevoicelistgcs( config.gcloud_token ).join(" ")
+            end
+
+          else
+            # auth
+            if ( match = message.match( /^auth (#{regexuser}) (#{regexb64})$/i ) )
+              pp match
+              remoteuser = match[1]
+              remotekey  = match[2]
+              if File.exists?(  config.statedir + "apikeys/" + remoteuser )
+                File.each_line( config.statedir + "apikeys/" + remoteuser ) do |localkey|
+                  puts "comparing #{localkey} to #{remotekey}"
+                  if ( localkey == remotekey )
+                    connections[client]["authed"] = "true"
+                    client.puts "authed #{ remoteuser }"
+                    connections[client]["user"] = remoteuser
+                    connections[client]["key"]  = remotekey
+                    break
+                  end
+                end
+              end
+              if ( connections[client]["authed"] == "false" )
+                client.puts "error: auth failure"
+              end
+            else
+              client.puts "must auth [user] [key]"
+            end
+          end
+        end
+        connections.delete( client )
+        channelsubs.each_key do |key|
+          channelsubs[key].delete( client )
+        end
+        puts "Disconnected: #{clientsocket.remote_address}"
+        pp channelsubs
+        pp connections
+      rescue ex : IO::Error
+        puts ex
+        next
+      rescue ex
+        puts ex
+        next
+      end
+    end
+  rescue ex
+    puts ex
+  end
+end
+