-- -- (C) 2014-26 - ntop.org -- -- This file contains a small set of utility functions for managing labels, e.g. aliases require("ntop_utils") -- ############################################## local function label2formattedlabel(alt_name, host_info, show_vlan, shorten_len) if not isEmptyString(alt_name) then local ip = host_info["ip"] or host_info["host"] -- Make it shorter local res = alt_name -- Special shorting function for IP addresses if res ~= ip then if (not shorten_len) or (shorten_len == false) then -- Don't touch the string, requested as-is without shortening elseif tonumber(shorten_len) then -- Shorten according to the specified length res = shortenString(res, shorten_len) else -- Use the default system-wide setting for the shortening res = shortenString(res) end end -- Adding the vlan if requested if show_vlan then local vlan = tonumber(host_info["vlan"]) if vlan and vlan > 0 then local full_vlan_name = getFullVlanName(vlan, true --[[ Compact --]]) res = string.format("%s@%s", res, full_vlan_name) end end return res end -- Fallback: just the IP and VLAN return (hostinfo2hostkey(host_info, true)) end -- ############################################## -- Flow Utils -- function flowinfo2hostname(flow_info, host_type, alerts_view, add_hostname) local name local orig_name require("check_redis_prefs") if (alerts_view and not hasClickHouseSupport()) or (add_hostname ~= nil and add_hostname == false) then -- do not return resolved name as it will hide the IP address return flow_info[host_type .. ".ip"] end if flow_info == nil then return "" end if host_type == "srv" then if flow_info["host_server_name"] ~= nil and flow_info["host_server_name"] ~= "" and flow_info["host_server_name"]:match("%w") and not isIPv4(flow_info["host_server_name"]) and not isIPv6(flow_info["host_server_name"]) then -- remove possible ports from the name return (flow_info["host_server_name"]:gsub(":%d+$", "")) end if flow_info["protos.tls.certificate"] ~= nil and flow_info["protos.tls.certificate"] ~= "" then return flow_info["protos.tls.certificate"] end end local hostinfo = { host = flow_info[host_type .. ".ip"], label = flow_info[host_type .. ".host"], -- mac = flow_info[host_type .. ".mac"], dhcpHost = flow_info[host_type .. ".dhcpHost"], broadcast_domain_host = flow_info[host_type .. ".broadcast_domain_host"], vlan = flow_info["vlan"], } mac = flow_info[host_type .. ".mac"] if (hostinfo.host == nil) and (not isEmptyString(mac)) then hostinfo.mac = mac end return (hostinfo2label(hostinfo, true, false, true)) end -- ############################################## -- Attempt at retrieving an host label from an host_info, using local caches and DNS resolution. -- This can be more expensive if compared to only using information found inside host_info. local function hostinfo2label_resolved(host_info, show_vlan, shorten_len, skip_resolution) local ip = host_info["ip"] or host_info["host"] local res if not (isEmptyString(ip)) then -- If local broadcast domain host and DHCP, try to get the label associated -- to the MAC address if host_info["mac"] and (host_info["broadcast_domain_host"] or host_info["dhcpHost"]) then res = getHostAltName(host_info["mac"]) end -- In case no resolution is requested, directly skip this part if (isEmptyString(res)) and not skip_resolution then -- Try and get the resolved name res = ntop.getResolvedName(ip) if (not isEmptyString(res)) and (res ~= ip) then res = string.lower(res) if host_info.names == nil then host_info.names = {} end host_info.names["DNS (Cache) Resolution"] = res interface.setHostResolvedName(ip, res) -- tprint(ip .. " = " .. res) else -- Nothing found, just fallback to the IP address res = ip end end else res = "Unknown" end return label2formattedlabel(res, host_info, show_vlan, shorten_len) end -- ############################################## -- "Some Very Long String" -> "Some Ver...g String" function shortenCollapse(s, max_len) local replacement = "..." local r_len = string.len(replacement) local s_len = string.len(s) if max_len == nil then max_len = ntop.getPref("ntopng.prefs.max_ui_strlen") max_len = tonumber(max_len) if max_len == nil then max_len = 24 end end if max_len <= r_len then return replacement end if s_len > max_len then local half = math.floor((max_len - r_len) / 2) return string.sub(s, 1, half) .. replacement .. string.sub(s, s_len - half + 1) end return s end -- ############################################## function getDhcpNameKey(ifid, mac) return string.format("ntopng.dhcp.%d.cache.%s", ifid, mac) end -- ############################################## function mac2label(mac) local alt_name = getHostAltName(mac) if not isEmptyString(alt_name) and (alt_name ~= mac) then return alt_name end alt_name = ntop.getCache(getDhcpNameKey(interface.getId(), mac)) if not isEmptyString(alt_name) and (alt_name ~= mac) then return alt_name end -- Fallback: just the MAC return mac end -- ############################################## -- Just a convenience function for hostinfo2label with only IP and VLAN function ip2label(ip, vlan, shorten_len) return hostinfo2label({ host = ip, vlan = (vlan or 0), }, true, shorten_len) end -- ############################################## -- URL Util -- -- -- Split the host key (ip@vlan) creating a new lua table. -- Note: removes /32 if cidr (ip/32@vlan) -- Example: -- info = hostkey2hostinfo(key) -- ip = info["host"] -- vlan = info["vlan"] -- function hostkey2hostinfo(key) local host = {} local info = split(key, "@") -- No @ if table.len(info) == 0 and not isEmptyString(key) then host["host"] = key end if info and info[1] ~= nil then host["host"] = info[1] end if info and info[2] ~= nil then host["vlan"] = tonumber(info[2]) else host["vlan"] = 0 end -- Handle CIDR (and remove /xx from host, if any) local info = split(host["host"], "/") if info and info[1] ~= nil then host["host"] = info[1] host["mask"] = tonumber(info[2]) end return host end -- ############################################## -- Split the pool key (ip/sub@vlan) creating a new lua table. -- Example: -- info = poolkey2poolinfo(key) -- ip = info["host"] -- vlan = info["vlan"] -- function poolkey2poolinfo(key) local pool = {} local info = split(key, "@") if table.len(info) == 0 and not isEmptyString(key) then pool["host"] = key end if info and info[1] ~= nil then pool["host"] = info[1] end if info and info[2] ~= nil then pool["vlan"] = tonumber(info[2]) else pool["vlan"] = 0 end return pool end -- ############################################## -- Retrieve an host label from an host_info. The minimum fields of -- the host_info are "host" and "vlan", however a full JSON from Host::lua -- is needed to provide an accurate result. -- -- The following order is used to determine the label: -- MAC label (LBD hosts only), IP label, MDNS/DHCP name from C, resolved IP -- -- NOTE: The function attempt at labelling an host only using information found in host_info. -- In case host_info is not enough to label the host, then local caches and DNS resolution kick in -- to find a label (at the expense of extra time). function hostinfo2label(host_info, show_vlan, shorten_len, skip_resolution) local ip = host_info["ip"] or host_info["host"] -- Take the label as found in the host structure local res = host_info.label if isEmptyString(res) then -- Use any user-configured custom name -- This goes first as a label set by the user MUST take precedance over any other possibly available label res = getHostAltName(ip) -- Prefer a user-configured device alias over names learned -- automatically from traffic for local L2/DHCP hosts. if isEmptyString(res) and host_info["mac"] and (host_info["broadcast_domain_host"] or host_info["dhcpHost"]) then res = getHostAltName(host_info["mac"]) end if isEmptyString(res) then -- Read what is found inside host `name`, e.g., name as found by dissected traffic such as DHCP res = host_info["name"] if isEmptyString(res) then return hostinfo2label_resolved(host_info, show_vlan, shorten_len, skip_resolution) end end end return label2formattedlabel(res, host_info, show_vlan, shorten_len) end -- ############################################## -- -- Analyze the host_info table and return the host key. -- Example: -- host_info = interface.getHostInfo("127.0.0.1",0) -- key = hostinfo2hostkey(host_info) -- function hostinfo2hostkey(host_info, host_type, show_vlan) local rsp = "" if host_type == "cli" then local cli_ip = host_info["cli.ip"] or host_info["cli_ip"] if cli_ip then rsp = rsp .. cli_ip end elseif host_type == "srv" then local srv_ip = host_info["srv.ip"] or host_info["srv_ip"] if srv_ip then rsp = rsp .. srv_ip end else if host_info["ip"] ~= nil then rsp = rsp .. host_info["ip"] elseif host_info["mac"] ~= nil then rsp = rsp .. host_info["mac"] elseif host_info["host"] ~= nil then rsp = rsp .. host_info["host"] elseif host_info["name"] ~= nil then rsp = rsp .. host_info["name"] end end local vlan_id = tonumber(host_info["vlan"] or host_info["vlan_id"] or 0) if vlan_id ~= 0 or show_vlan then rsp = rsp .. "@" .. tostring(vlan_id) end return rsp end -- ############################################## function getHostAltNamesKey(host_key) if host_key == nil then return nil end return "ntopng.cache.host_labels." .. host_key end -- ############################################## function getHostAltName(host_info) local host_key local alt_name -- Check if there is an alias for the host@vlan -- Note: this is not used for backward compatibility (see setHostAltName) -- if type(host_info) == "table" and host_info["vlan"] then -- host_key = hostinfo2hostkey(host_info) -- alt_name = ntop.getCache(getHostAltNamesKey(host_key)) -- return alt_name -- end -- Check if there is an alias for the host if type(host_info) == "table" then host_key = host_info["host"] else host_key = host_info end local k = getHostAltNamesKey(host_key) if k ~= nil then alt_name = ntop.getCache(k) or "" -- In case the alt_name and the key are the same, return nil if (not isEmptyString(alt_name)) and (string.lower(alt_name) == string.lower(host_key)) then return nil end else alt_name = "" end return alt_name end -- ############################################## function setHostAltName(host_info, alt_name) local host_key if type(host_info) == "table" then -- Note: we are not using hostinfo2hostkey which includes the -- vlan for backward compatibility, compatibility with -- the backend, and compatibility with the vpn scripts host_key = host_info["host"] -- hostinfo2hostkey(host_info) else host_key = host_info end local key = getHostAltNamesKey(host_key) if isEmptyString(alt_name) then ntop.delCache(key) else ntop.setCache(key, alt_name) end end -- ############################################## -- This function set the interface alias, return true if the -- alias is set, false otherwise function setInterfaceAlias(iface, alias) local ok = true if isEmptyString(iface) then ok = false end if ok and (iface ~= alias) and not isEmptyString(alias) then ntop.setCache(getInterfaceAliasKey(iface), alias) else ok = false end return ok end -- ############################################## function getLocalNetworkAliasKey() return "ntopng.network_aliases" end -- ############################################## function getInterfaceAliasKey(ifid) return "ntopng.prefs.ifid_" .. ifid .. ".name" end -- ############################################## function getInterfaceAlias(ifid) if not ifid or not tonumber(ifid) then return "" end return ntop.getCache("ntopng.prefs.ifid_" .. tostring(ifid) .. ".name") or "" end -- ############################################## -- network_id -> network_key of the networks of an interface, built once per -- interface: interface.getNetworksStats() returns the stats of all the networks -- and calling it for each lookup was quadratic (e.g. ~1.3s with ~350 networks) local network_keys_by_ifid = {} local function getNetworkKeysById() local ifid = interface.getId() local keys = network_keys_by_ifid[ifid] if not keys then keys = {} for _, ns in pairs(interface.getNetworksStats() or {}) do -- Same as the original linear scan: the last match wins keys[ns.network_id] = ns.network_key end network_keys_by_ifid[ifid] = keys end return keys end local function computeLocalNetworkAliasById(network, remove_slash) local network_consts = require("network_consts") local network_id = tonumber(network) -- If network is unknown (u_int32_t max) then return an empty value if network_id == nil or network_id >= network_consts.UNKNOWN_NETWORK then return " " end local network_key = getNetworkKeysById()[network_id] if network_key then return getFullLocalNetworkName(network_key, remove_slash) end return "" end function getLocalNetworkAliasById(network, remove_slash) return computeLocalNetworkAliasById(network, remove_slash) end -- ############################################## -- All the network aliases set by the user, read with a single Redis call the -- first time an alias is needed (it was one call per network, slow with many -- networks, e.g. building the network filter options) local local_network_aliases = nil -- Updates the cached alias of `network` (see setLocalNetworkAlias) function setCachedLocalNetworkAlias(network, alias) if local_network_aliases then local_network_aliases[network] = alias end end function getLocalNetworkAlias(network) local alias = "" -- Try with the cache on redis if not local_network_aliases then local_network_aliases = ntop.getHashAllCache(getLocalNetworkAliasKey()) or {} end alias = local_network_aliases[network] -- Try with the C++ naming (config file) if isEmptyString(alias) then alias = ntop.getLocalNetworkAlias(network) end -- All fail, return the network if isEmptyString(alias) then return network end -- Alias is not empty, parse for _ and substitute with spaces alias = string.gsub(alias, "_", " ") return alias end -- ############################################## function getLocalNetworkLabel(network) local alias = getLocalNetworkAlias(network) if alias ~= network then return string.format("%s", alias) end return network end -- ############################################## function getFullLocalNetworkName(network, remove_slash) local network_consts = require("network_consts") local network_id = tonumber(network) local alias = getLocalNetworkAlias(network) if network_id ~= nil and network_id >= network_consts.UNKNOWN_NETWORK then return "" end if alias ~= network then if remove_slash then local items = split(alias, "/") alias = items[#items] end return string.format("%s ยท %s", alias, network) end return network end -- ############################################## function getVlanAliasKey() return "ntopng.vlan_aliases" end -- ############################################## function getVlanAlias(vlan_id) local alias = ntop.getHashCache(getVlanAliasKey(), vlan_id) if not isEmptyString(alias) then return alias end return tostring(vlan_id) end -- ############################################## local vlan_cached_list = {} function getFullVlanName(vlan_id, compact, return_untagged) -- In case of vlan 0, return empty string as name -- fix for untagged vlan (#7998) -- Also do not check redis in case of vlan 0 if (tonumber(vlan_id) == 0) or (vlan_id == nil) then if return_untagged then return i18n("no_vlan") end return "" end local alias = "" if not vlan_cached_list[vlan_id] then vlan_cached_list[vlan_id] = getVlanAlias(vlan_id) or "" end alias = vlan_cached_list[vlan_id] if not isEmptyString(alias) then if not isEmptyString(alias) and alias ~= tostring(vlan_id) then if compact then alias = shortenString(alias) return string.format("%s", alias) else return string.format("%u [%s]", vlan_id, alias) end end end return vlan_id end -- ################################# -- This function actively resolves an host if there is not information about it. -- NOTE: prefer the host2name on this function function resolveAddress(hostinfo, allow_empty, shorten_len) local alt_name = ip2label(hostinfo["host"], hostinfo["vlan"], shorten_len) if not isEmptyString(alt_name) and (alt_name ~= hostinfo["host"]) then -- The host label has priority return alt_name end local hostname = ntop.resolveName(hostinfo["host"]) if isEmptyString(hostname) then -- Not resolved if allow_empty == true then return hostname else -- this function will take care of formatting the IP return hostinfo2label(hostinfo, true, shorten_len) end end return hostinfo2label(hostinfo, true, shorten_len) end -- ############################################## -- See also getHumanReadableInterfaceName function getInterfaceName(interface_id, windows_skip_description) if interface_id == getSystemInterfaceId() then return (getSystemInterfaceName()) end local ifnames = interface.getIfNames() local iface = ifnames[tostring(interface_id)] if iface ~= nil then if windows_skip_description ~= true and string.contains(iface, "{") then -- Windows local old_iface = interface.getId() -- Use the interface description instead of the name interface.select(tostring(iface)) iface = interface.getStats().description interface.select(tostring(old_iface)) end return iface end return "" end -- ############################################## function getHumanReadableInterfaceName(interface_name, ignore_custom_name) local interface_id = nil if interface_name == "__system__" then return (i18n("system")) elseif tonumber(interface_name) ~= nil then -- convert ID to name interface_id = tonumber(interface_name) interface_name = getInterfaceName(interface_name) else -- Parameter is a string, let's take it's id first interface_id = getInterfaceId(interface_name) -- and then get the name interface_name = getInterfaceName(interface_id) end if not ignore_custom_name then local custom_name = getInterfaceAlias(interface_id) if not isEmptyString(custom_name) then return (shortenCollapse(custom_name)) end end return interface_name end