src/zm_stream_socket.cpp
Rama master · réplica de lectura del repositorio original.
/*
* This file is part of the ZoneMinder Project. See AUTHORS file for Copyright information
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* Free Software Foundation; either version 2 of the License, or (at your
* option) any later version.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
* more details.
*
* You should have received a copy of the GNU General Public License along
* with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "zm_stream_socket.h"
#include "zm_config.h"
#include "zm_logger.h"
#include "zm_utils.h"
#include <algorithm>
#include <cerrno>
#include <cinttypes>
#include <cstring>
#include <limits>
#include <grp.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/uio.h>
#include <sys/un.h>
#include <unistd.h>
using namespace zm::stream_socket;
StreamSocket::StreamSocket(unsigned int monitor_id, std::string path) :
StreamSocket(monitor_id, std::move(path), Config()) {
}
StreamSocket::StreamSocket(unsigned int monitor_id, std::string path, Config config) :
monitor_id_(monitor_id),
path_(std::move(path)),
config_(std::move(config)) {
}
StreamSocket::~StreamSocket() {
Stop();
}
bool StreamSocket::Start() {
if (thread_.joinable())
return true;
// zm::UnixSocket copies the path with a truncating strncpy, so an over-long
// path would bind one file while the chmod, chown and unlink below act on
// another. Refuse it instead.
if (path_.size() >= sizeof(sockaddr_un::sun_path)) {
Error("StreamSocket: socket path %s is longer than the %zu bytes a unix socket"
" address allows", path_.c_str(), sizeof(sockaddr_un::sun_path) - 1);
return false;
}
::unlink(path_.c_str());
if (!listener_.bind(path_.c_str())) {
Error("StreamSocket: failed to bind %s", path_.c_str());
listener_.close();
return false;
}
ApplyPermissions();
if (!listener_.listen()) {
Error("StreamSocket: failed to listen on %s", path_.c_str());
listener_.close();
::unlink(path_.c_str());
return false;
}
listener_.setBlocking(false);
if (!wake_.open()) {
Error("StreamSocket: failed to open wake pipe");
listener_.close();
::unlink(path_.c_str());
return false;
}
wake_.setBlocking(false);
terminate_ = false;
thread_ = std::thread(&StreamSocket::Run, this);
Debug(1, "StreamSocket: monitor %u listening on %s", monitor_id_, path_.c_str());
return true;
}
void StreamSocket::Stop() {
if (thread_.joinable()) {
terminate_ = true;
Wake();
thread_.join();
}
if (listener_.isOpen()) {
listener_.close();
::unlink(path_.c_str());
}
wake_.close();
}
void StreamSocket::ApplyPermissions() {
if (chmod(path_.c_str(), 0660) < 0)
Warning("StreamSocket: chmod 0660 %s failed: %s", path_.c_str(), strerror(errno));
if (config_.group.empty())
return;
long size = sysconf(_SC_GETGR_R_SIZE_MAX);
std::vector<char> buffer(size > 0 ? size : 1024);
group grp = {};
group *result = nullptr;
if (getgrnam_r(config_.group.c_str(), &grp, buffer.data(), buffer.size(), &result) != 0
or !result) {
Warning("StreamSocket: group %s not found, leaving default group on %s",
config_.group.c_str(), path_.c_str());
return;
}
if (chown(path_.c_str(), static_cast<uid_t>(-1), grp.gr_gid) < 0)
Warning("StreamSocket: chown group %s on %s failed: %s",
config_.group.c_str(), path_.c_str(), strerror(errno));
}
StreamSocket::Config StreamSocket::ConfigFromStatic() {
Config config;
config.group = staticConfig.STREAM_SOCKET_GROUP;
config.allowed_uids = ParseAllowedUids(staticConfig.STREAM_SOCKET_ALLOWED_UIDS);
if (staticConfig.STREAM_SOCKET_MAX_CLIENTS > 0)
config.max_clients = staticConfig.STREAM_SOCKET_MAX_CLIENTS;
if (staticConfig.STREAM_SOCKET_QUEUE_BYTES > 0)
config.queue_max_bytes = staticConfig.STREAM_SOCKET_QUEUE_BYTES;
if (staticConfig.STREAM_SOCKET_QUEUE_MSGS > 0)
config.queue_max_msgs = staticConfig.STREAM_SOCKET_QUEUE_MSGS;
if (staticConfig.STREAM_SOCKET_STALL_SECS > 0)
config.stall_timeout = std::chrono::seconds(staticConfig.STREAM_SOCKET_STALL_SECS);
return config;
}
std::vector<uid_t> StreamSocket::ParseAllowedUids(const std::string &value) {
std::vector<uid_t> uids;
for (const std::string &token : Split(value, ',')) {
std::string trimmed = Trim(token, " \t");
if (trimmed.empty())
continue;
errno = 0;
char *end = nullptr;
unsigned long uid = strtoul(trimmed.c_str(), &end, 10);
// strtoul wraps a leading '-' and saturates on overflow; reject both, and
// any value that does not round-trip through uid_t, so a typo cannot
// silently allow a different uid (e.g. 2^32 truncating to uid 0).
bool out_of_range = errno == ERANGE
or uid != static_cast<unsigned long>(static_cast<uid_t>(uid));
if (trimmed[0] != '-' and end != trimmed.c_str() and *end == '\0' and !out_of_range) {
uids.push_back(static_cast<uid_t>(uid));
} else {
Warning("StreamSocket: ignoring malformed uid '%s' in allowed uids", trimmed.c_str());
}
}
return uids;
}
bool StreamSocket::CheckPeer(int fd, uid_t &uid, pid_t &pid) const {
// Kernel-verified peer credentials: SO_PEERCRED on Linux, getpeereid on the
// BSDs and macOS (which report no pid). Without either the uid allow-list
// cannot be enforced and only the filesystem permissions apply.
#if defined(SO_PEERCRED)
ucred cred = {};
socklen_t len = sizeof(cred);
if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &cred, &len) != 0) {
Warning("StreamSocket: SO_PEERCRED failed on %s: %s", path_.c_str(), strerror(errno));
return config_.allowed_uids.empty();
}
uid = cred.uid;
pid = cred.pid;
#elif defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
gid_t gid = 0;
if (getpeereid(fd, &uid, &gid) != 0) {
Warning("StreamSocket: getpeereid failed on %s: %s", path_.c_str(), strerror(errno));
return config_.allowed_uids.empty();
}
pid = 0;
#else
if (!config_.allowed_uids.empty()) {
Warning("StreamSocket: peer credentials unavailable on this platform,"
" ZM_STREAM_SOCKET_ALLOWED_UIDS is not enforced on %s", path_.c_str());
}
return true;
#endif
if (config_.allowed_uids.empty() or uid == geteuid())
return true;
return std::find(config_.allowed_uids.begin(), config_.allowed_uids.end(), uid)
!= config_.allowed_uids.end();
}
namespace {
// HELLO body for a stream, or empty when the source has no such stream.
std::vector<uint8_t> HelloPayload(const AVCodecParameters *par, AVRational frame_rate) {
if (!par or par->codec_id == AV_CODEC_ID_NONE)
return {};
return BuildHello(par, frame_rate);
}
} // namespace
void StreamSocket::SetStreams(const AVCodecParameters *video, AVRational frame_rate,
const AVCodecParameters *audio) {
std::vector<uint8_t> video_payload = HelloPayload(video, frame_rate);
std::vector<uint8_t> audio_payload = HelloPayload(audio, {0, 0});
bool changed;
{
std::lock_guard<std::mutex> lock(mutex_);
changed = ApplyStreamsLocked(std::move(video_payload), std::move(audio_payload));
}
if (changed)
Wake();
}
void StreamSocket::SetVideoParams(const AVCodecParameters *par, AVRational frame_rate) {
std::vector<uint8_t> payload = HelloPayload(par, frame_rate);
bool changed;
{
std::lock_guard<std::mutex> lock(mutex_);
changed = ApplyStreamsLocked(std::move(payload), std::vector<uint8_t>(hello_audio_payload_));
}
if (changed)
Wake();
}
void StreamSocket::SetAudioParams(const AVCodecParameters *par) {
std::vector<uint8_t> payload = HelloPayload(par, {0, 0});
bool changed;
{
std::lock_guard<std::mutex> lock(mutex_);
changed = ApplyStreamsLocked(std::vector<uint8_t>(hello_video_payload_), std::move(payload));
}
if (changed)
Wake();
}
void StreamSocket::ClearVideoParams() {
SetVideoParams(nullptr, {0, 0});
}
void StreamSocket::ClearAudioParams() {
SetAudioParams(nullptr);
}
bool StreamSocket::ApplyStreamsLocked(std::vector<uint8_t> video, std::vector<uint8_t> audio) {
bool video_changed = video != hello_video_payload_;
bool audio_changed = audio != hello_audio_payload_;
if (!video_changed and !audio_changed)
return false;
// A generation is complete once its video HELLO has gone out; from then on
// any change - new parameters, a stream appearing, a stream disappearing -
// starts a new one. Before that (the initial audio-then-video announcement)
// only a change to an already announced stream does.
bool bump = !hello_video_payload_.empty()
or (audio_changed and !hello_audio_payload_.empty());
if (bump) {
++generation_;
sequence_[0] = sequence_[1] = 0;
keyframe_.reset(); // its header carries the old generation
Info("StreamSocket: monitor %u stream parameters changed (video %s, audio %s),"
" generation now %u", monitor_id_,
video.empty() ? "none" : (video_changed ? "changed" : "same"),
audio.empty() ? "none" : (audio_changed ? "changed" : "same"),
generation_);
}
hello_video_payload_ = std::move(video);
hello_audio_payload_ = std::move(audio);
// On a bump every remaining stream is re-announced under the new generation;
// otherwise only what is new. Audio first: the video HELLO is always the
// last HELLO of a generation.
if (hello_audio_payload_.empty()) {
hello_audio_.reset();
} else if (bump or audio_changed) {
hello_audio_ = MakeMessage(MessageType::Hello, StreamId::Audio, 0, 0, 0,
std::vector<uint8_t>(hello_audio_payload_), true);
BroadcastLocked(hello_audio_);
}
if (hello_video_payload_.empty()) {
hello_video_.reset();
} else if (bump or video_changed) {
hello_video_ = MakeMessage(MessageType::Hello, StreamId::Video, 0, 0, 0,
std::vector<uint8_t>(hello_video_payload_), true);
BroadcastLocked(hello_video_);
}
return true;
}
void StreamSocket::SendMedia(const AVPacket *packet, StreamId stream,
bool keyframe, int64_t pts_us) {
if (!packet or packet->size <= 0)
return;
// Only the two media streams have a sequence counter and a HELLO
if (stream != StreamId::Video and stream != StreamId::Audio)
return;
bool video_keyframe = keyframe and stream == StreamId::Video;
std::unique_lock<std::mutex> lock(mutex_);
// A stream is on the wire only once its parameters are announced. Without a
// HELLO a consumer cannot decode the payload (no codec id or extradata), so
// dropping here keeps the protocol's "HELLO precedes MEDIA" guarantee for
// e.g. audio packets a monitor forwards before record_audio announces them.
const std::vector<uint8_t> &hello =
stream == StreamId::Audio ? hello_audio_payload_ : hello_video_payload_;
if (hello.empty())
return;
uint32_t sequence = sequence_[static_cast<size_t>(stream)]++;
bool have_clients = !clients_.empty();
// With no consumers the message only counts against the sequence; skip the
// packet reference and fan-out entirely. Keyframes are still cached so the
// first consumer to connect gets an immediate first frame.
if (!have_clients and !video_keyframe)
return;
Header header = {};
header.length = kHeaderLengthBytes + packet->size;
header.version = kProtocolVersion;
header.type = static_cast<uint8_t>(MessageType::Media);
header.stream = static_cast<uint8_t>(stream);
header.flags = video_keyframe ? kFlagKeyframe : 0;
header.sequence = sequence;
header.generation = generation_;
header.pts_us = pts_us; // Header.pts_us is signed
if (video_keyframe) {
// Cache the keyframe for fast-start of late joiners; the cache references
// the packet's buffer, no copy is made.
av_packet_ptr cache_clone{av_packet_clone(packet)};
if (cache_clone) {
auto cache = std::make_shared<Message>();
Header cache_header = header;
cache_header.type = static_cast<uint8_t>(MessageType::Keyframe);
SerializeHeader(cache_header, cache->header.data());
cache->av_payload = std::move(cache_clone);
keyframe_ = std::move(cache);
}
}
if (!have_clients)
return;
// Reference the payload buffer; no copy of the media data is made.
av_packet_ptr clone{av_packet_clone(packet)};
if (!clone) {
Error("StreamSocket: av_packet_clone failed for monitor %u", monitor_id_);
return;
}
auto message = std::make_shared<Message>();
SerializeHeader(header, message->header.data());
message->av_payload = std::move(clone);
BroadcastLocked(message);
lock.unlock();
Wake();
}
void StreamSocket::SendMonitorEvent(std::vector<uint8_t> payload) {
std::unique_lock<std::mutex> lock(mutex_);
uint32_t sequence = event_sequence_++;
if (clients_.empty())
return; // sequence still advanced, so a late joiner sees the gap
MessagePtr message = MakeMessage(MessageType::Event, StreamId::Monitor, 0,
sequence, 0, std::move(payload), true);
BroadcastLocked(message);
lock.unlock();
Wake();
}
void StreamSocket::SetSnapshotEvent(std::vector<uint8_t> payload) {
std::lock_guard<std::mutex> lock(mutex_);
// The snapshot is the consumer's authoritative current status on connect. It
// is not broadcast, and only the body is kept: AcceptClient frames it, so
// the header carries the generation and the event-sequence baseline that are
// current when the consumer connects rather than when the status last moved.
snapshot_payload_ = std::move(payload);
}
void StreamSocket::InvalidateKeyframe() {
std::lock_guard<std::mutex> lock(mutex_);
keyframe_.reset();
}
StreamSocket::MessagePtr StreamSocket::MakeMessage(
MessageType type, StreamId stream, uint8_t flags, uint32_t sequence,
int64_t pts_us, std::vector<uint8_t> payload, bool control) const {
auto message = std::make_shared<Message>();
Header header = {};
header.length = kHeaderLengthBytes + payload.size();
header.version = kProtocolVersion;
header.type = static_cast<uint8_t>(type);
header.stream = static_cast<uint8_t>(stream);
header.flags = flags;
header.sequence = sequence;
header.generation = generation_;
header.pts_us = pts_us; // Header.pts_us is signed
SerializeHeader(header, message->header.data());
message->blob_payload = std::move(payload);
message->control = control;
return message;
}
void StreamSocket::BroadcastLocked(const MessagePtr &message) {
for (std::unique_ptr<Client> &client : clients_)
EnqueueLocked(*client, message);
}
void StreamSocket::EnqueueLocked(Client &client, const MessagePtr &message) {
while (!client.queue.empty()
and (client.queued_bytes + message->size() > config_.queue_max_bytes
or client.queue.size() + 1 > config_.queue_max_msgs)) {
// Drop the oldest non-control message. The front message is unsafe to
// drop once partially written (removing it would desync the framing) or
// while the listener is writing it with the lock released.
bool front_pinned = client.front_offset > 0 or client.front_in_flight;
auto begin = client.queue.begin() + (front_pinned ? 1 : 0);
auto victim = std::find_if(begin, client.queue.end(),
[](const MessagePtr &m) { return !m->control; });
if (victim == client.queue.end())
break;
client.queued_bytes -= (*victim)->size();
client.queue.erase(victim);
++client.dropped;
}
if (client.queued_bytes + message->size() > config_.queue_max_bytes
or client.queue.size() + 1 > config_.queue_max_msgs) {
if (!message->control) {
// Nothing droppable left; drop the incoming message instead.
++client.dropped;
return;
}
// Control messages (HELLO/BYE) always go through, even above the limit.
}
client.queue.push_back(message);
client.queued_bytes += message->size();
}
bool StreamSocket::DrainClient(Client &client) {
while (true) {
// Peek the front message under the lock and pin it: EnqueueLocked may
// drop queued messages to make room while we are writing, but never the
// one in flight, so the bookkeeping below always refers to it.
MessagePtr front;
size_t offset = 0;
{
std::lock_guard<std::mutex> lock(mutex_);
if (client.queue.empty())
return true;
front = client.queue.front();
offset = client.front_offset;
client.front_in_flight = true;
}
const Message &message = *front;
iovec iov[2];
int iovcnt = 0;
if (offset < kHeaderSize) {
iov[iovcnt].iov_base = const_cast<uint8_t *>(message.header.data()) + offset;
iov[iovcnt].iov_len = kHeaderSize - offset;
++iovcnt;
if (message.payload_size() > 0) {
iov[iovcnt].iov_base = const_cast<uint8_t *>(message.payload_data());
iov[iovcnt].iov_len = message.payload_size();
++iovcnt;
}
} else {
size_t payload_offset = offset - kHeaderSize;
iov[iovcnt].iov_base = const_cast<uint8_t *>(message.payload_data()) + payload_offset;
iov[iovcnt].iov_len = message.payload_size() - payload_offset;
++iovcnt;
}
msghdr msg = {};
msg.msg_iov = iov;
msg.msg_iovlen = iovcnt;
// MSG_NOSIGNAL: a disconnected client must produce EPIPE, not SIGPIPE.
// This is the kernel copy of the payload; it runs unlocked.
ssize_t written = ::sendmsg(client.sock->getDesc(), &msg, MSG_NOSIGNAL);
int error = written < 0 ? errno : 0;
std::lock_guard<std::mutex> lock(mutex_);
client.front_in_flight = false;
if (written < 0) {
if (error == EAGAIN or error == EWOULDBLOCK)
return true;
if (error == EINTR)
continue;
Debug(1, "StreamSocket: sendmsg to client pid %d failed: %s",
client.pid, strerror(error));
return false;
}
client.front_offset += written;
client.last_progress = std::chrono::steady_clock::now();
if (client.front_offset >= message.size()) {
client.queued_bytes -= message.size();
client.queue.pop_front();
client.front_offset = 0;
++client.sent;
}
}
}
void StreamSocket::SendStatsLocked(Client &client, TimePoint now) {
if (now - client.last_stats < config_.stats_interval)
return;
client.last_stats = now;
MessagePtr stats = MakeMessage(MessageType::Stats, StreamId::Video, 0, 0, 0,
BuildStats(client.sent, client.dropped), false);
EnqueueLocked(client, stats);
}
void StreamSocket::AcceptClient() {
while (true) {
int fd = ::accept(listener_.getDesc(), nullptr, nullptr);
if (fd < 0) {
if (errno != EAGAIN and errno != EWOULDBLOCK and errno != EINTR)
Warning("StreamSocket: accept on %s failed: %s", path_.c_str(), strerror(errno));
return;
}
auto sock = std::make_unique<zm::TcpUnixSocket>(listener_, fd);
uid_t uid = 0;
pid_t pid = 0;
if (!CheckPeer(fd, uid, pid)) {
Warning("StreamSocket: rejecting client uid %u pid %d on %s (not in allowed uids)",
uid, pid, path_.c_str());
continue; // sock destructor closes the connection
}
std::lock_guard<std::mutex> lock(mutex_);
if (clients_.size() >= config_.max_clients) {
Warning("StreamSocket: rejecting client uid %u pid %d on %s (%zu clients connected)",
uid, pid, path_.c_str(), clients_.size());
continue;
}
sock->setBlocking(false);
auto client = std::make_unique<Client>();
client->sock = std::move(sock);
client->uid = uid;
client->pid = pid;
client->last_progress = std::chrono::steady_clock::now();
client->last_stats = client->last_progress;
// New consumers get the stream parameters first (audio HELLO before video
// HELLO, the video HELLO always completes a generation's set), then the
// current-status snapshot, then the cached keyframe so they can render
// immediately instead of waiting for the next GOP.
if (hello_audio_)
EnqueueLocked(*client, hello_audio_);
if (hello_video_)
EnqueueLocked(*client, hello_video_);
if (!snapshot_payload_.empty()) {
EnqueueLocked(*client, MakeMessage(MessageType::Event, StreamId::Monitor, 0,
event_sequence_, 0,
std::vector<uint8_t>(snapshot_payload_), true));
}
if (keyframe_)
EnqueueLocked(*client, keyframe_);
Info("StreamSocket: monitor %u client connected uid %u pid %d (%zu clients)",
monitor_id_, uid, pid, clients_.size() + 1);
clients_.push_back(std::move(client));
}
}
void StreamSocket::Wake() {
if (wake_.isOpen()) {
uint8_t byte = 0;
if (::write(wake_.getWriteDesc(), &byte, 1) < 0 and errno != EAGAIN) {
Debug(1, "StreamSocket: wake write failed: %s", strerror(errno));
}
}
}
void StreamSocket::Run() {
std::vector<pollfd> fds;
while (!terminate_) {
fds.clear();
fds.push_back({listener_.getDesc(), POLLIN, 0});
fds.push_back({wake_.getReadDesc(), POLLIN, 0});
{
std::lock_guard<std::mutex> lock(mutex_);
for (std::unique_ptr<Client> &client : clients_) {
short events = POLLIN;
if (!client->queue.empty())
events |= POLLOUT;
fds.push_back({client->sock->getDesc(), events, 0});
}
}
// With no clients there is nothing to drain or report; sleep until a
// connection or a wake. Idle monitors cost no periodic wakeups.
int timeout = fds.size() > 2 ? 100 : -1;
int ready = ::poll(fds.data(), fds.size(), timeout);
if (ready < 0) {
if (errno == EINTR)
continue;
Error("StreamSocket: poll failed on %s: %s", path_.c_str(), strerror(errno));
break;
}
if (terminate_)
break;
if (fds[1].revents & POLLIN) {
uint8_t buffer[64];
while (::read(wake_.getReadDesc(), buffer, sizeof(buffer)) > 0) {}
}
if (fds[0].revents & POLLIN)
AcceptClient();
// clients_ and each Client's socket are owned by this thread; only the
// queues and counters are shared with producers, so the lock is taken
// just around those, never around the socket writes.
TimePoint now = std::chrono::steady_clock::now();
size_t fd_index = 2;
for (auto it = clients_.begin(); it != clients_.end();) {
Client &client = **it;
short revents = 0;
if (fd_index < fds.size() and fds[fd_index].fd == client.sock->getDesc()) {
revents = fds[fd_index].revents;
++fd_index;
}
bool remove = false;
if (revents & (POLLERR | POLLHUP | POLLNVAL))
remove = true;
if (!remove and (revents & POLLIN)) {
// Protocol v1 has no client->server messages; discard inbound bytes,
// a zero read means the peer closed.
uint8_t buffer[256];
ssize_t bytes = ::recv(client.sock->getDesc(), buffer, sizeof(buffer), 0);
if (bytes == 0)
remove = true;
}
if (!remove)
remove = !DrainClient(client);
if (!remove) {
std::lock_guard<std::mutex> lock(mutex_);
SendStatsLocked(client, now);
if (client.queued_bytes > 0 and now - client.last_progress > config_.stall_timeout) {
Warning("StreamSocket: monitor %u disconnecting stalled client uid %u pid %d"
" (%zu bytes queued, %" PRIu64 " dropped)",
monitor_id_, client.uid, client.pid, client.queued_bytes, client.dropped);
remove = true;
}
}
if (remove) {
std::lock_guard<std::mutex> lock(mutex_);
Info("StreamSocket: monitor %u client uid %u pid %d disconnected"
" (sent %" PRIu64 ", dropped %" PRIu64 ")",
monitor_id_, client.uid, client.pid, client.sent, client.dropped);
it = clients_.erase(it);
} else {
++it;
}
}
}
// Best-effort BYE so consumers can tell shutdown from failure
{
std::lock_guard<std::mutex> lock(mutex_);
MessagePtr bye = MakeMessage(MessageType::Bye, StreamId::Video, 0, 0, 0, {}, true);
for (std::unique_ptr<Client> &client : clients_)
EnqueueLocked(*client, bye);
}
for (std::unique_ptr<Client> &client : clients_)
DrainClient(*client);
std::lock_guard<std::mutex> lock(mutex_);
clients_.clear();
}