winamp/Src/Plugins/DSP/sc_serv3/protocol_shoutcastSource.cpp
2024-09-24 14:54:57 +02:00

439 lines
15 KiB
C++

#ifdef _WIN32
#include <winsock2.h>
#endif
#include "protocol_shoutcastSource.h"
#include "protocol_backup.h"
#include "streamData.h"
#include "global.h"
#include "bandwidth.h"
#include "MP3Header.h"
#include "ADTSHeader.h"
#include "services/stdServiceImpl.h"
using namespace std;
using namespace uniString;
using namespace stringUtil;
#define DEBUG_LOG(...) do { if (gOptions.shoutcastSourceDebug()) DLOG(__VA_ARGS__); } while (0)
#define LOGNAME "SRC"
protocol_shoutcastSource::protocol_shoutcastSource (microConnection &mc, const uniString::utf8 &password) throw (std::exception)
: runnable (mc), m_srcPort(mc.m_srcPort), m_srcAddr(mc.m_srcAddress)
{
m_srcStreamID = DEFAULT_SOURCE_STREAM;
m_denied = false;
m_remainder = new __uint8[BUF_SIZE * 4];
m_remainderSize = 0;
m_outBuffer = NULL;
m_outBufferSize = 0;
m_streamData = NULL;
DEBUG_LOG(m_srcLogString + __FUNCTION__);
// we're looking to see if this is an updated 1.x source
// which is able to indicate the stream # for the stream
// so that we're able to support multiple 1.x sources so
// we need to parse the password and extract the parts
utf8 m_srcPassword = password;
extractPassword(m_srcPassword, m_srcUserID, m_srcStreamID);
// ensure that only valid stream id's are allowed to connect (1 -> 2147483647)
if (!m_srcStreamID || (m_srcStreamID > INT_MAX))
{
m_denied = true;
ELOG(m_srcLogString + "Bad Stream ID (" + tos(m_srcStreamID) + "). Stream ID cannot be below 1 or above 2147483647.");
m_outBuffer = MSG_BADSTREAMID;
bandWidth::updateAmount(bandWidth::SOURCE_V1_SENT, (m_outBufferSize = MSG_BADSTREAMID_LEN));
m_state = &protocol_shoutcastSource::state_SendBuffer;
m_nextState = &protocol_shoutcastSource::state_CloseConnection;
return;
}
// update the log message for the read stream number
m_srcLogString = srcAddrLogString (m_srcAddr, m_srcPort, m_srcStreamID);
// if we have a moved stream then now we have the stream id
// then we need to check and block the source as applicable
utf8 movedUrl = gOptions.stream_movedUrl(m_srcStreamID);
if (!movedUrl.empty())
{
m_denied = true;
ELOG(m_srcLogString + "Shoutcast 1 source rejected. Stream is configured as having moved.");
m_outBuffer = MSG_STREAMMOVED;
bandWidth::updateAmount(bandWidth::SOURCE_V1_SENT, (m_outBufferSize = MSG_STREAMMOVED_LEN));
m_state = &protocol_shoutcastSource::state_SendBuffer;
m_nextState = &protocol_shoutcastSource::state_CloseConnection;
return;
}
// as we are a v1 source then we must adhere to the master password
// instead of using a specific per stream password as in v2 streams
// though we also accept connections as sid=1 so check for that too
utf8 srcPassword = gOptions.stream_password(m_srcStreamID);
if (srcPassword.empty())
{
srcPassword = gOptions.password();
}
if (m_srcPassword.empty() || (m_srcPassword != srcPassword))
{
m_denied = true;
ELOG(m_srcLogString + "Shoutcast 1 source connection denied" + (m_srcUserID.empty() ? "" : " for user (" + m_srcUserID + ")") +
". " + (m_srcPassword.empty() ? "Empty password not allowed." : "Bad password: " + m_srcPassword), LOGNAME, m_srcStreamID);
m_outBuffer = MSG_INVALIDPASSWORD;
bandWidth::updateAmount(bandWidth::SOURCE_V1_SENT, (m_outBufferSize = MSG_INVALIDPASSWORD_LEN));
m_state = &protocol_shoutcastSource::state_SendBuffer;
m_nextState = &protocol_shoutcastSource::state_CloseConnection;
}
else
{
// if we've got a source already connected and it's not a backup
// then it's better that we just abort processing now than later
bool isSourceActive = false;
streamData *sd = streamData::accessStream(m_srcStreamID, isSourceActive);
if (sd && (isSourceActive == true) && (sd->isBackupStream(m_srcStreamID) == false))
{
m_denied = true;
ELOG(m_srcLogString + "Shoutcast 1 source rejected. A source is already connected.", LOGNAME, m_srcStreamID);
m_outBuffer = MSG_STREAMINUSE;
bandWidth::updateAmount(bandWidth::SOURCE_V1_SENT, (m_outBufferSize = MSG_STREAMINUSE_LEN));
m_state = &protocol_shoutcastSource::state_SendBuffer;
m_nextState = &protocol_shoutcastSource::state_CloseConnection;
}
else
{
ILOG(m_srcLogString + "Shoutcast 1 source connection starting.", LOGNAME, m_srcStreamID);
m_outBuffer = MSG_VALIDPASSWORD;
bandWidth::updateAmount(bandWidth::SOURCE_V1_SENT, (m_outBufferSize = MSG_VALIDPASSWORD_LEN));
m_state = &protocol_shoutcastSource::state_SendBuffer;
m_nextState = &protocol_shoutcastSource::state_GetHeaders;
}
if (sd)
{
sd->releaseStream();
}
}
}
protocol_shoutcastSource::~protocol_shoutcastSource() throw()
{
DEBUG_LOG(m_srcLogString + __FUNCTION__);
if (m_streamData)
{
streamData::streamSourceLost(m_srcLogString, m_streamData, m_srcStreamID);
m_streamData = 0;
}
socketOps::forgetTCPSocket(m_socket);
forgetArray(m_remainder);
if (!m_denied)
{
ILOG(m_srcLogString + "Shoutcast 1 source disconnected.", LOGNAME, m_srcStreamID);
}
}
void protocol_shoutcastSource::state_AnalyzeHeaders() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__);
int maxHeaderLineCount = gOptions.maxHeaderLineCount();
if ((int)m_headers.size() >= maxHeaderLineCount)
{
m_denied = true;
throwEx<runtime_error>(m_srcLogString + "Max icy header lines exceeded");
}
m_lineBuffer = stripWhitespace(m_lineBuffer);
if (m_lineBuffer.empty())
{
// adjust icy headers for titleFormat and urlFormat
utf8 titleFormat = gOptions.titleFormat();
utf8 urlFormat = gOptions.urlFormat();
if (!titleFormat.empty())
{
utf8::size_type pos = titleFormat.find(utf8("%s"));
m_headers["icy-name"] = (pos == utf8::npos ? titleFormat : titleFormat.replace(pos,2,m_headers["icy-name"]));
}
if (!urlFormat.empty())
{
utf8::size_type pos = urlFormat.find(utf8("%s"));
m_headers["icy-url"] = (pos == utf8::npos ? urlFormat : urlFormat.replace(pos,2,m_headers["icy-url"]));
}
// dump icy headers to log
if (gOptions.shoutcastSourceDebug())
{
for (map<utf8,utf8>::const_iterator i = m_headers.begin(); i != m_headers.end(); ++i)
{
DEBUG_LOG(m_srcLogString + "Source client header [" + (*i).first + ":" + (*i).second + "]");
}
}
config::streamConfig stream;
const bool found = gOptions.getStreamConfig(stream, m_srcStreamID);
if (!found && gOptions.requireStreamConfigs())
{
m_denied = true;
throwEx<runtime_error>(m_srcLogString + "Shoutcast 1 source rejected. Stream " +
tos(m_srcStreamID) + " must be defined in config file");
}
// check that these bitrates are allowed (looking at both max and average values)
const int bitrate = getStreamBitrate(m_headers) * 1000;
int streamMaxBitrate = 0, streamMinBitrate = 0;
const int ret = gOptions.isBitrateDisallowed(m_srcStreamID, bitrate, streamMaxBitrate, streamMinBitrate);
if (ret)
{
m_denied = true;
utf8 mode = ((streamMaxBitrate == streamMinBitrate) ? "of" : (ret == 2 ? "up to" : "from"));
throwEx<runtime_error>(m_srcLogString + "Shoutcast 1 source rejected. Only bitrates " +
mode + " " + tos((ret == 1 ? streamMinBitrate : streamMaxBitrate) / 1000) +
" kbps are allowed - detected " + tos(bitrate / 1000) + " kbps.");
}
m_streamData = streamData::createStream(streamData::streamSetup(m_srcLogString, m_srcAddr,
(found ? stream.m_authHash : ""), m_srcUserID, "",
stream.m_backupUrl.url(), streamData::SHOUTCAST1,
m_srcStreamID, m_srcPort, stream.m_maxStreamUser,
stream.m_maxStreamBitrate, stream.m_minStreamBitrate,
stream.m_allowPublicRelay, false, getStreamSamplerate(m_headers),
mapGet(m_headers, "icy-vbr", (bool)false), m_headers));
if (!m_streamData)
{
m_denied = true;
ELOG(m_srcLogString + "Shoutcast 1 source rejected. A source is already connected.");
m_outBuffer = MSG_STREAMINUSE;
bandWidth::updateAmount(bandWidth::SOURCE_V1_SENT, (m_outBufferSize = MSG_STREAMINUSE_LEN));
m_state = &protocol_shoutcastSource::state_SendBuffer;
m_nextState = &protocol_shoutcastSource::state_CloseConnection;
m_result.run();
return;
}
utf8 sourceIdent = mapGet(m_headers, "user-agent", utf8());
m_streamData->updateSourceIdent(sourceIdent);
m_state = &protocol_shoutcastSource::state_GetStreamData;
m_result.read();
}
else
{
// find the colon that divides header lines into key/value fields
utf8::size_type pos = m_lineBuffer.find(utf8(":"));
utf8 key = toLower(stripWhitespace(m_lineBuffer.substr(0, pos)));
if (pos == utf8::npos)
{
if (!key.empty() && ((key == "icy-name") || (key == "icy-url")))
{
// allow through icy-name and icy-url if there is
// a titleformat and urlformat to use respectively
}
else
{
m_denied = true;
throwEx<runtime_error>(m_srcLogString + "Shoutcast 1 source connection rejected. "
"Bad icy header string [" + m_lineBuffer + "]");
}
}
utf8 value = stripWhitespace(m_lineBuffer.substr(pos+1));
if (key.empty() || value.empty())
{
if (!key.empty() && value.empty())
{
if (key == "icy-genre")
{
value = "Misc";
}
else if (((key == "icy-name") && !gOptions.titleFormat().empty()) ||
((key == "icy-url") && !gOptions.urlFormat().empty()))
{
// allow through icy-name and icy-url if there is
// a titleformat and urlformat to use respectively
}
else
{
if (key == "icy-url")
{
value = "http://www.shoutcast.com";
}
else if (!((key == "icy-irc") || (key == "icy-aim") || (key == "icy-icq")))
{
m_denied = true;
throwEx<runtime_error>(m_srcLogString + "Shoutcast 1 source connection rejected. "
"Bad icy header string [" + m_lineBuffer + "]");
}
}
}
else
{
m_denied = true;
throwEx<runtime_error>(m_srcLogString + "Shoutcast 1 source connection rejected. "
"Bad icy header string [" + m_lineBuffer + "]");
}
}
m_headers[key] = value;
m_nextState = &protocol_shoutcastSource::state_AnalyzeHeaders;
m_state = &protocol_shoutcastSource::state_GetLine;
m_result.read();
m_lineBuffer.clear();
}
}
void protocol_shoutcastSource::timeSlice() throw(std::exception)
{
try
{
if (m_streamData && m_streamData->isDead())
{
m_result.done();
return;
}
(this->*m_state)();
}
catch(const exception &)
{
if (m_streamData && !m_denied)
{
// if there was a failure, now see if we have a backup and attempt to run
// before we remove the current handling of the dropped source connection
vector<config::streamConfig> backupInfo = gOptions.getBackupUrl(m_srcStreamID);
if (!backupInfo.empty())
{
streamData::streamInfo info;
streamData::extraInfo extra;
if (streamData::getStreamInfo (m_streamData->ID(), info, extra) && info.m_allowBackupURL)
{
m_denied = true;
m_streamData->clearCachedMetadata();
streamData::streamSourceLost(m_srcLogString, m_streamData, m_srcStreamID);
m_streamData = 0;
ILOG (m_srcLogString + "Shoutcast 1 source disconnected - trying source backup.", LOGNAME, m_srcStreamID);
threadedRunner::scheduleRunnable(new protocol_backup(backupInfo[0], getStreamBitrate(m_headers),
fixMimeType(m_headers["content-type"])));
}
else
WLOG ("Stream backup URL not allowed", LOGNAME, m_srcStreamID);
}
}
throw;
}
}
void protocol_shoutcastSource::state_SendBuffer() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__);
if (sendDataBuffer(m_srcStreamID, m_outBuffer, m_outBufferSize, m_srcLogString))
{
m_state = m_nextState;
}
}
void protocol_shoutcastSource::state_GetLine() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__);
if (getHTTPStyleHeaderLine(m_srcStreamID, m_lineBuffer, m_srcLogString))
{
m_state = m_nextState;
}
}
void protocol_shoutcastSource::state_GetHeaders() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__);
m_state = &protocol_shoutcastSource::state_GetLine;
m_nextState = &protocol_shoutcastSource::state_AnalyzeHeaders;
m_result.read();
}
void protocol_shoutcastSource::state_GetStreamData() throw(exception)
{
/*#if defined(_DEBUG) || defined(DEBUG)
DEBUG_LOG(m_srcLogString + __FUNCTION__);
#endif*/
time_t cur_time;
const int autoDumpTime = detectAutoDumpTimeout (cur_time, m_srcStreamID, (m_srcLogString + "Timeout waiting for stream data"));
int bitrate = m_streamData->streamBitrate();
const int type = m_streamData->streamUvoxDataType();
while (true)
{
char buf[BUF_SIZE * 4] = {0};
int amt = (BUF_SIZE - 1);
// if we had anything left over then now we
// need to copy it back into the buffer and
// adjust the max data amount to be read in
if ((m_remainderSize > 0) && ((amt + m_remainderSize) <= (BUF_SIZE * 4)))
{
memcpy(buf, m_remainder, m_remainderSize);
}
else
{
m_remainderSize = 0;
}
// adjust the position in the buffer based on the prior
// state of the remaining data as part of frame syncing
int rval = 0;
if ((rval = recv (&buf[m_remainderSize], (BUF_SIZE - 1), 0x0)) < 1)
{
if (rval < 0)
{
rval = socketOps::errCode();
if (rval == SOCKETOPS_WOULDBLOCK)
{
m_result.schedule(85);
m_result.timeout((autoDumpTime - (int)(cur_time - m_lastActivityTime)));
return;
}
DLOG (m_srcLogString + "Socket error while waiting for data. " + socketErrString(rval), LOGNAME, m_srcStreamID);
}
else
DLOG (m_srcLogString + "Remote socket closed while waiting for data.", LOGNAME, m_srcStreamID);
throwEx<runtime_error> ((utf8)"");
}
// update these details before we mess with anything
bandWidth::updateAmount(bandWidth::SOURCE_V1_RECV, rval);
// if we're here then we account for what we already had in the total
// so that we then don't skip the new data read with the original data
rval += m_remainderSize;
m_remainderSize = 0;
amt = rval;
if (m_streamData->syncToStream(m_remainderSize, m_remainder, amt,
bitrate, type, buf, m_srcLogString))
{
m_denied = true;
throwEx<runtime_error>(m_srcLogString + "Shoutcast 1 source disconnected. "
"Unable to sync to the stream. Please check the "
"source is valid and in a supported format.");
}
m_lastActivityTime = ::time(NULL);
m_result.schedule(25);
}
}
void protocol_shoutcastSource::state_CloseConnection() throw(exception)
{
DEBUG_LOG(m_srcLogString + __FUNCTION__);
m_result.done();
}