mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-30 00:27:11 +09:00
479 lines
20 KiB
Python
479 lines
20 KiB
Python
"""The panel half of bounded log collection.
|
|
|
|
What matters: a file is followed from where it was left and rotation starts
|
|
a new generation; ids are stable so a resend is a duplicate, not a second
|
|
line; every bound Cloud enforces is applied here first; Cloud's answer is the
|
|
configuration and `collect: false` stops collection at once; nothing that
|
|
looks like a secret crosses the relay.
|
|
"""
|
|
import json
|
|
import os
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.services import connect_logs
|
|
|
|
|
|
@pytest.fixture()
|
|
def logfile(tmp_path):
|
|
path = tmp_path / 'serverkit.log'
|
|
path.write_text('')
|
|
return path
|
|
|
|
|
|
def test_a_file_is_followed_from_where_it_was_left_and_rotation_is_a_generation(logfile):
|
|
tail = connect_logs.FileTail('serverkit.service', str(logfile))
|
|
# First sight starts at the end: history is not what Cloud keeps.
|
|
logfile.write_text('old line\n', newline='\n')
|
|
assert tail.read() == []
|
|
|
|
with open(logfile, 'a', encoding='utf-8', newline='\n') as fh:
|
|
fh.write('INFO one\nERROR two\npartial')
|
|
got = tail.read()
|
|
assert [r['text'] for r in got] == ['INFO one', 'ERROR two']
|
|
assert got[0]['severity'] == 'info' and got[1]['severity'] == 'error'
|
|
assert got[0]['generation'] == 0 and got[0]['offset'] == len('old line\n')
|
|
ids = {r['id'] for r in got}
|
|
|
|
# The partial line arrives once its newline does, and nothing is resent.
|
|
with open(logfile, 'a', encoding='utf-8', newline='\n') as fh:
|
|
fh.write(' finished\n')
|
|
again = tail.read()
|
|
assert [r['text'] for r in again] == ['partial finished']
|
|
assert not ids & {r['id'] for r in again}
|
|
|
|
# Truncation is a rotation: generation moves on, the offset restarts.
|
|
logfile.write_text('fresh\n', newline='\n')
|
|
rotated = tail.read()
|
|
assert rotated[0]['generation'] == 1 and rotated[0]['offset'] == 0
|
|
|
|
|
|
def test_ids_are_stable_and_lines_are_bounded_and_redacted(logfile):
|
|
assert connect_logs.record_id('a', 1, 2) == connect_logs.record_id('a', 1, 2)
|
|
assert connect_logs.record_id('a', 1, 2) != connect_logs.record_id('a', 2, 2)
|
|
|
|
tail = connect_logs.FileTail('web.error', str(logfile))
|
|
tail.read()
|
|
with open(logfile, 'a', encoding='utf-8', newline='\n') as fh:
|
|
fh.write('token=ghp_abcdefghijklmnop123456 password: hunter2 '
|
|
'https://u:p@example.test/ Authorization: Bearer abcdefghijkl\n')
|
|
fh.write('x' * (connect_logs.MAX_LINE_BYTES + 100) + '\n')
|
|
got = tail.read()
|
|
assert 'hunter2' not in got[0]['text'] and 'ghp_' not in got[0]['text']
|
|
assert 'u:p@' not in got[0]['text'] and 'abcdefghijkl' not in got[0]['text']
|
|
assert len(got[1]['text'].encode()) <= connect_logs.MAX_LINE_BYTES
|
|
assert got[1]['truncated'] is True
|
|
text = json.dumps(got)
|
|
assert str(logfile) not in text, 'a path never crosses the relay'
|
|
|
|
|
|
def test_batches_respect_every_bound(monkeypatch):
|
|
pub = connect_logs.LogPublisher(None)
|
|
pub.sources = []
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['serverkit.service'],
|
|
'flush_interval_s': 5})
|
|
for i in range(450):
|
|
pub.pending.append({'id': f'r{i}', 'source': 'serverkit.service',
|
|
'observed_at': 'now', 'text': 'x' * 900})
|
|
pub.pending_bytes = sum(len(json.dumps(r).encode()) for r in pub.pending)
|
|
first = pub.take_batch()
|
|
assert first['schema'] == 'logs/1'
|
|
assert len(first['records']) <= connect_logs.MAX_BATCH_RECORDS
|
|
assert len(json.dumps(first).encode()) <= connect_logs.MAX_BATCH_BYTES
|
|
# Flushes are spaced; the next one is not due yet.
|
|
assert pub.due() is False
|
|
pub.next_flush_at = 0.0
|
|
second = pub.take_batch()
|
|
assert second['generation'] == first['generation'] + 1
|
|
assert first['records'][0]['id'] != second['records'][0]['id']
|
|
|
|
|
|
@pytest.mark.parametrize('key', [
|
|
'password', 'api_key', 'API-KEY', 'token', 'authorization', 'private_key',
|
|
])
|
|
def test_json_secrets_are_redacted_without_changing_other_fields(key):
|
|
secret = 'private "quoted" value\\with\\slashes'
|
|
line = json.dumps({
|
|
'message': 'request failed',
|
|
'context': [{key: secret, 'status': 401}],
|
|
})
|
|
redacted = json.loads(connect_logs.redact(line))
|
|
assert redacted == {
|
|
'message': 'request failed',
|
|
'context': [{key: '[redacted]', 'status': 401}],
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize('secret', [123456, True, None])
|
|
def test_json_secret_scalars_remain_valid_json(secret):
|
|
line = json.dumps({'password': secret, 'status': 'failed'})
|
|
assert json.loads(connect_logs.redact(line)) == {
|
|
'password': '[redacted]', 'status': 'failed',
|
|
}
|
|
|
|
|
|
def test_prefixed_json_is_redacted_before_a_file_record_is_published(logfile):
|
|
tail = connect_logs.FileTail('serverkit.service', str(logfile))
|
|
tail.read()
|
|
line = 'ERROR request ' + json.dumps({
|
|
'password': 'example-secret', 'api_key': 'example-key', 'status': 401,
|
|
})
|
|
logfile.write_text(line + '\n', newline='\n')
|
|
records = tail.read()
|
|
assert len(records) == 1
|
|
assert json.loads(records[0]['text'].removeprefix('ERROR request ')) == {
|
|
'password': '[redacted]', 'api_key': '[redacted]', 'status': 401,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize('extra_bytes', [0, 1, connect_logs.MAX_READ_BYTES + 20])
|
|
def test_oversized_lines_do_not_block_later_records(logfile, extra_bytes):
|
|
tail = connect_logs.FileTail('serverkit.service', str(logfile))
|
|
tail.read()
|
|
prefix = b'INFO before\n'
|
|
oversized = b'password=' + b'x' * (
|
|
connect_logs.MAX_READ_BYTES + extra_bytes - len(b'password='))
|
|
logfile.write_bytes(prefix + oversized + b'\nINFO after\n')
|
|
|
|
records = []
|
|
for _ in range(8):
|
|
records.extend(tail.read())
|
|
assert [r['text'] for r in records] == [
|
|
'INFO before', '[oversized log line omitted]', 'INFO after',
|
|
]
|
|
assert records[1]['truncated'] is True
|
|
assert records[1]['offset'] == len(prefix)
|
|
assert records[2]['offset'] == len(prefix) + len(oversized) + 1
|
|
assert records[1]['id'] == connect_logs.record_id(
|
|
'serverkit.service', 0, len(prefix))
|
|
assert tail.offset == logfile.stat().st_size
|
|
assert tail.read() == []
|
|
|
|
|
|
def test_oversized_unterminated_line_waits_for_newline_without_leaking_remainder(logfile):
|
|
tail = connect_logs.FileTail('serverkit.service', str(logfile))
|
|
tail.read()
|
|
logfile.write_bytes(b'password=' + b'x' * connect_logs.MAX_READ_BYTES)
|
|
assert tail.read()[0]['truncated'] is True
|
|
assert tail.read() == []
|
|
assert tail.offset == logfile.stat().st_size
|
|
|
|
with logfile.open('ab') as fh:
|
|
fh.write(b'secret-continuation')
|
|
assert tail.read() == []
|
|
with logfile.open('ab') as fh:
|
|
fh.write(b'\nINFO recovered\n')
|
|
assert [r['text'] for r in tail.read()] == ['INFO recovered']
|
|
|
|
|
|
@pytest.mark.parametrize('rotate', [False, True])
|
|
def test_rotation_resets_oversized_line_discarding(logfile, rotate):
|
|
tail = connect_logs.FileTail('serverkit.service', str(logfile))
|
|
tail.read()
|
|
logfile.write_bytes(b'x' * connect_logs.MAX_READ_BYTES)
|
|
assert tail.read()[0]['truncated'] is True
|
|
if rotate:
|
|
logfile.rename(logfile.with_suffix('.old'))
|
|
logfile.write_bytes(b'INFO fresh file\n')
|
|
records = tail.read()
|
|
assert [r['text'] for r in records] == ['INFO fresh file']
|
|
assert records[0]['generation'] == 1
|
|
assert records[0]['offset'] == 0
|
|
|
|
|
|
def test_a_refused_batch_goes_back_to_the_front_and_a_stop_clears_it():
|
|
pub = connect_logs.LogPublisher(None)
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
pub.pending = [{'id': 'b', 'source': 'web.error', 'observed_at': 'now', 'text': 'b'}]
|
|
batch = {'schema': 'logs/1', 'generation': 1,
|
|
'records': [{'id': 'a', 'source': 'web.error', 'observed_at': 'now', 'text': 'a'}]}
|
|
pub.requeue(batch)
|
|
assert [r['id'] for r in pub.pending] == ['a', 'b']
|
|
|
|
# Consent withdrawn: stop now, hold nothing, ask again only slowly.
|
|
pub.apply_ack({'ok': False, 'reason': 'scope_not_granted', 'collect': False, 'sources': []})
|
|
assert pub.collect is False and pub.pending == [] and pub.pending_bytes == 0
|
|
assert pub.due() is False
|
|
pub.retry_after_refusal()
|
|
assert pub.collect is False, 'not before the retry window'
|
|
pub.refused_until = 0.0
|
|
pub.retry_after_refusal()
|
|
assert pub.collect is True and pub.sources is None
|
|
assert pub.due() is True, 'one empty batch asks for the configuration'
|
|
|
|
|
|
def test_before_cloud_answers_nothing_is_read_and_the_first_batch_is_empty(monkeypatch):
|
|
pub = connect_logs.LogPublisher(None)
|
|
assert pub.wanted() == []
|
|
assert pub.collect_now() == 0
|
|
assert pub.due() is True
|
|
batch = pub.take_batch()
|
|
assert batch['records'] == []
|
|
# Cloud names the sources; only those are read from then on.
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error', 'not-a-source']})
|
|
assert pub.wanted() == ['web.error']
|
|
|
|
|
|
def test_deploy_output_is_sent_once_per_deployment(tmp_path):
|
|
log = tmp_path / 'build-7.json'
|
|
log.write_text(json.dumps({'logs': [{'message': 'Step 1'}, {'message': 'token=abc123456789'},
|
|
{'message': ''}]}))
|
|
|
|
class Dep:
|
|
id = 7
|
|
app_id = 3
|
|
version = 2
|
|
build_log_path = str(log)
|
|
|
|
reader = connect_logs.DeployOutput(None)
|
|
reader._recent = lambda: [Dep()]
|
|
got = reader.read()
|
|
assert [r['text'] for r in got] == ['Step 1', 'token=[redacted]']
|
|
assert got[0]['app_key'] == '3' and got[0]['deployment_key'] == '3:2'
|
|
assert reader.read() == [], 'once'
|
|
assert str(log) not in json.dumps(got)
|
|
|
|
|
|
def test_the_client_wires_the_stream_and_requeues_an_unacknowledged_batch():
|
|
from app.services import connect_client
|
|
client = connect_client.RelayClient(None)
|
|
client._logs.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
client._logs.pending = [{'id': 'a', 'source': 'web.error', 'observed_at': 'now', 'text': 'a'}]
|
|
client._logs.pending_bytes = 60
|
|
sent = []
|
|
|
|
class WS:
|
|
def send(self, raw):
|
|
sent.append(json.loads(raw))
|
|
assert client._publish_logs(WS()) is True
|
|
assert sent[0]['k'] == 'logs' and sent[0]['p']['records'][0]['id'] == 'a'
|
|
stream_id = sent[0]['s']
|
|
assert stream_id in client._logs_inflight
|
|
|
|
# A refusal that is not a stop: the batch goes back.
|
|
client._handle_frame(WS(), json.dumps({'s': stream_id, 't': 'close',
|
|
'p': {'ok': False, 'reason': 'malformed',
|
|
'collect': True, 'sources': ['web.error']}}))
|
|
assert [r['id'] for r in client._logs.pending] == ['a']
|
|
assert stream_id not in client._logs_inflight
|
|
|
|
# A stop: nothing is held and nothing more is sent.
|
|
client._logs.next_flush_at = 0.0
|
|
client._publish_logs(WS())
|
|
stream_id = sent[-1]['s']
|
|
client._handle_frame(WS(), json.dumps({'s': stream_id, 't': 'close',
|
|
'p': {'ok': False, 'reason': 'scope_not_granted',
|
|
'collect': False, 'sources': []}}))
|
|
assert client._logs.collect is False and client._logs.pending == []
|
|
assert client._publish_logs(WS()) is False
|
|
|
|
|
|
@pytest.mark.parametrize('line', [
|
|
'Authorization: Bearer abc', 'Authorization: Basic x',
|
|
'client_secret=hidden refresh_token=hidden',
|
|
'{"client_secret":"hidden","refresh_token":"hidden"}',
|
|
"password='hidden\\'also-hidden'",
|
|
])
|
|
def test_short_credentials_and_compound_keys_are_redacted(line):
|
|
redacted = connect_logs.redact(line)
|
|
assert 'hidden' not in redacted
|
|
assert not redacted.endswith((' abc', ' x'))
|
|
assert '[redacted]' in redacted
|
|
|
|
|
|
def test_file_tail_rejects_symlinks(logfile, tmp_path):
|
|
tail = connect_logs.FileTail('web.error', str(logfile))
|
|
tail.read()
|
|
target = tmp_path / 'private.txt'
|
|
target.write_text('private file contents\n')
|
|
logfile.unlink()
|
|
try:
|
|
logfile.symlink_to(target)
|
|
except OSError:
|
|
pytest.skip('symlinks require privileges on this platform')
|
|
assert tail.read() == []
|
|
assert tail.offset == 0
|
|
|
|
|
|
def test_replacement_between_stat_and_open_is_not_read(logfile, monkeypatch):
|
|
tail = connect_logs.FileTail('web.error', str(logfile))
|
|
tail.read()
|
|
original_open = os.open
|
|
|
|
def replace_then_open(path, flags):
|
|
logfile.rename(logfile.with_suffix('.old'))
|
|
logfile.write_bytes(b'private replacement contents\n')
|
|
return original_open(path, flags)
|
|
|
|
monkeypatch.setattr(connect_logs.os, 'open', replace_then_open)
|
|
assert tail.read() == []
|
|
assert tail.offset == 0
|
|
|
|
|
|
def test_panel_reader_uses_production_error_log(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(connect_logs.paths, 'SERVERKIT_LOG_DIR', str(tmp_path))
|
|
assert connect_logs._panel_log_path() == str(tmp_path / 'error.log')
|
|
(tmp_path / 'error.log').write_text('')
|
|
(tmp_path / 'serverkit.log').write_text('')
|
|
assert connect_logs._panel_log_path() == str(tmp_path / 'error.log')
|
|
|
|
|
|
def test_deployment_reader_drains_more_than_one_batch_and_new_lines(tmp_path):
|
|
path = tmp_path / 'build.json'
|
|
lines = [f'line {i}' for i in range(450)]
|
|
path.write_text(json.dumps({'logs': lines}))
|
|
reader = connect_logs.DeployOutput()
|
|
reader._recent = lambda: [
|
|
SimpleNamespace(id=7, app_id=3, version=2, build_log_path=str(path))]
|
|
records = reader.read() + reader.read() + reader.read()
|
|
assert [r['text'] for r in records] == lines
|
|
assert len({r['id'] for r in records}) == 450
|
|
assert reader.read() == []
|
|
path.write_text(json.dumps({'logs': lines + ['last line']}))
|
|
assert [r['text'] for r in reader.read()] == ['last line']
|
|
|
|
|
|
def test_repeated_app_messages_are_distinct_and_overlaps_are_skipped(monkeypatch):
|
|
from app.services.log_service import LogService
|
|
reader = connect_logs.AppOutput()
|
|
reader._applications = lambda: [
|
|
SimpleNamespace(id=7, name='web', app_type='docker', compose_file=None)]
|
|
first = 'web | 2026-09-11T01:00:00.123456789Z ERROR repeated'
|
|
later = 'web | 2026-09-11T01:00:01.123456789Z ERROR repeated'
|
|
snapshots = iter([[first, first], [first, first], [first, later]])
|
|
|
|
def read_logs(*args, **kwargs):
|
|
assert kwargs['timestamps'] is True
|
|
return {'success': True, 'lines': next(snapshots)}
|
|
|
|
monkeypatch.setattr(LogService, 'get_docker_app_logs', read_logs)
|
|
records = reader.read()
|
|
assert len(records) == 2
|
|
assert records[0]['id'] != records[1]['id']
|
|
assert reader.read() == []
|
|
newer = reader.read()
|
|
assert len(newer) == 1 and newer[0]['text'] == later
|
|
assert newer[0]['id'] not in {r['id'] for r in records}
|
|
|
|
|
|
@pytest.mark.parametrize('consent', [None, False, 1, 'true'])
|
|
def test_log_consent_requires_literal_true(consent):
|
|
pub = connect_logs.LogPublisher()
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
payload = {'ok': True}
|
|
if consent is not None:
|
|
payload['collect'] = consent
|
|
pub.apply_ack(payload)
|
|
assert pub.collect is False
|
|
assert pub.collect_now() == 0
|
|
assert pub.pending == []
|
|
|
|
|
|
def test_requeue_obeys_byte_bound_and_preserves_original_expiry(monkeypatch):
|
|
clock = [100.0]
|
|
monkeypatch.setattr(connect_logs.time, 'monotonic', lambda: clock[0])
|
|
pub = connect_logs.LogPublisher()
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
record = lambda key: {'id': key, 'source': 'web.error', 'text': 'x' * 100}
|
|
size = len(json.dumps(record('a')).encode())
|
|
monkeypatch.setattr(connect_logs, 'OUTAGE_BUFFER_BYTES', size * 2)
|
|
pub.pending = [record('a'), record('b')]
|
|
pub.pending_bytes = size * 2
|
|
pub.oldest_pending_at = clock[0]
|
|
batch = pub.take_batch()
|
|
assert '_queued_at' not in connect_logs.logs_frame('log1', batch)['p']
|
|
pub.pending = [record('c')]
|
|
pub.pending_bytes = size
|
|
pub.oldest_pending_at = 200.0
|
|
clock[0] = 200.0
|
|
pub.requeue(batch)
|
|
assert pub.pending_bytes <= size * 2
|
|
assert [r['id'] for r in pub.pending] == ['a', 'b']
|
|
retry = pub.take_batch()
|
|
clock[0] = 401.0
|
|
pub.requeue(retry)
|
|
assert pub.pending == []
|
|
|
|
|
|
def test_restricted_source_selection_purges_queued_records():
|
|
pub = connect_logs.LogPublisher()
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error', 'app.output']})
|
|
pub.pending = [
|
|
{'id': 'a', 'source': 'app.output', 'text': 'app'},
|
|
{'id': 'b', 'source': 'web.error', 'text': 'web'},
|
|
]
|
|
pub.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
assert [r['id'] for r in pub.pending] == ['b']
|
|
pub.requeue({'records': [{'id': 'a', 'source': 'app.output', 'text': 'app'}]})
|
|
assert [r['id'] for r in pub.pending] == ['b']
|
|
|
|
|
|
def test_log_ack_timeout_retries_only_after_empty_consent_probe(monkeypatch):
|
|
from app.services import connect_client
|
|
clock = [100.0]
|
|
monkeypatch.setattr(connect_logs.time, 'monotonic', lambda: clock[0])
|
|
client = connect_client.RelayClient()
|
|
client._logs.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
client._logs._readers['web.error'] = SimpleNamespace(read=lambda: [])
|
|
client._logs.pending = [{'id': 'a', 'source': 'web.error', 'text': 'message'}]
|
|
client._logs.pending_bytes = 60
|
|
client._logs.oldest_pending_at = clock[0]
|
|
frames = []
|
|
ws = SimpleNamespace(send=lambda raw: frames.append(json.loads(raw)))
|
|
assert client._publish_logs(ws)
|
|
assert not client._publish_logs(ws)
|
|
clock[0] = 161.0
|
|
assert client._publish_logs(ws)
|
|
assert frames[-1]['p']['records'] == []
|
|
assert [r['id'] for r in client._logs.pending] == ['a']
|
|
assert len(client._logs_inflight) == 1
|
|
client._handle_frame(ws, json.dumps({
|
|
's': frames[-1]['s'], 't': 'close',
|
|
'p': {'ok': True, 'collect': True, 'sources': ['web.error']},
|
|
}))
|
|
clock[0] = 167.0
|
|
assert client._publish_logs(ws)
|
|
assert frames[-1]['p']['records'][0]['id'] == 'a'
|
|
|
|
|
|
def test_socket_drop_recovers_unacknowledged_logs(monkeypatch):
|
|
from app.services import connect_client
|
|
client = connect_client.RelayClient()
|
|
client.running = True
|
|
client._logs.apply_ack({'ok': True, 'collect': True, 'sources': ['web.error']})
|
|
batch = {'schema': 'logs/1', 'generation': 1,
|
|
'records': [{'id': 'a', 'source': 'web.error', 'text': 'message'}],
|
|
'_queued_at': connect_logs.time.monotonic()}
|
|
client._logs_inflight['log1'] = batch
|
|
|
|
def drop(**kwargs):
|
|
raise ConnectionError('connection dropped before ack')
|
|
|
|
ws = SimpleNamespace(recv=drop, close=lambda: None)
|
|
monkeypatch.setattr(client, '_ws_connect', lambda cfg: ws)
|
|
monkeypatch.setattr(client, '_set_state', lambda *args, **kwargs: None)
|
|
for name in ('_check_for_update', '_load_jwks', '_publish_storage',
|
|
'_publish_policy', '_publish_inventory'):
|
|
monkeypatch.setattr(client, name, lambda *args: None)
|
|
assert client._ws_session({}).startswith('drop:')
|
|
assert client._logs_inflight == {}
|
|
assert client._logs.pending == batch['records']
|
|
assert client._logs.sources is None
|
|
assert client._logs.take_batch()['records'] == []
|
|
|
|
|
|
def test_timestamped_log_service_options_reach_commands(fake_subprocess, monkeypatch):
|
|
from app.services import log_service
|
|
fake_subprocess.script(['docker', 'compose', 'logs'], stdout='timestamp message\n')
|
|
result = log_service.LogService.get_docker_app_logs(
|
|
'web', '.', timestamps=True)
|
|
assert result['success']
|
|
assert '--timestamps' in fake_subprocess.argv_for(['docker', 'compose'])
|
|
|
|
monkeypatch.setattr(log_service, 'is_command_available', lambda name: True)
|
|
fake_subprocess.script(['journalctl'], stdout='timestamp message\n')
|
|
result = log_service.LogService.get_journalctl_logs(unit='web.service', precise=True)
|
|
assert result['success']
|
|
assert 'short-iso-precise' in fake_subprocess.argv_for(['journalctl'])
|