Sumber yang diterbitkan

CH-J Server Manager

Jelajahi direktori dan file untuk rilis aplikasi tertentu.

Unduh ZIP sumber

CH-J / Server Manager 377 file

core-20261019T111616Z-160a00bfcddc Ditambahkan: 19 Okt 2026, 22.11 · Dibuat: 19 Okt 2026, 13.16

remoteEditorLinux.py Ditambahkan: Tanggal tidak tercatat 19,3 KB · 392 baris

Kode Unduh file

1"""Exercise the shipped helper against real Linux descriptors and metadata.

3Only account-home lookup is redirected into an isolated test directory. No SSH,

4sudo, or external server is involved in this filesystem integration fixture.

5"""

6import base64

7import contextlib

8import errno

9import hashlib

10import io

11import json

12import os

13import pwd

14import stat

15import struct

16import sys

17import tempfile

18from types import SimpleNamespace

19from unittest.mock import patch

21source = open(sys.argv[1], encoding='utf-8').read()

22scenario = sys.argv[2]

24test_parent = pwd.getpwuid(os.getuid()).pw_dir if os.geteuid() == 0 else None

25with tempfile.TemporaryDirectory(prefix='chj-editor-test-', dir=test_parent) as root, contextlib.ExitStack() as stack:

26 home = root + '/actual home'

27 if scenario == 'cross-filesystem-strategy':

28 if not os.path.isdir('/dev/shm') or os.stat('/dev/shm').st_dev == os.stat(root).st_dev:

29 print('No separate writable filesystem available')

30 sys.exit(77)

        home = stack.enter_context(tempfile.TemporaryDirectory(prefix='chj-editor-home-', dir='/dev/shm')) + '/actual home'

32 os.mkdir(home, 0o700)

33 parent = root + '/target directory'

34 os.mkdir(parent, 0o700)

35 path = parent + '/config\' ; $(touch unsafe) `test` ü'

36 original = b'original\n'

37 modified = b'modified\n'

38 with open(path, 'wb') as out:

39 out.write(original)

40 os.chmod(path, 0o6750)

42 def command(request):

43 captured = io.StringIO()

        with patch.object(pwd, 'getpwuid', return_value=SimpleNamespace(pw_dir=home)), patch.object(sys, 'argv', ['remoteEditor.py', json.dumps(request)]), contextlib.redirect_stdout(captured):

45 exec(compile(source, 'remoteEditor.py', 'exec'), {})

46 return json.loads(captured.getvalue())

48 def value(request):

49 result = command(request)

50 assert result['ok'], result

51 return result['value']

53 baseline = value(dict(operation='read', path=path))['baseline']

54 ident = 'a' * 32

55 digest = hashlib.sha256(modified).hexdigest()

56 request = dict(operation='finalize', id=ident, uid=os.getuid(), path=path, baseline=baseline, hash=digest)

58 def stage(data=modified):

59 prepared = value(dict(operation='prepare', id=ident, path=path, baseline=request['baseline'], hash=digest))

60 fd = os.open(prepared['temporary'], os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

61 with os.fdopen(fd, 'wb') as out:

62 out.write(data)

63 assert stat.S_IMODE(os.stat(home + '/ch-j-sm').st_mode) == 0o700

64 assert stat.S_IMODE(os.stat(prepared['temporary']).st_mode) == 0o600

65 return prepared['temporary']

67 def intact(staged):

68 assert open(path, 'rb').read() == original

69 assert open(staged, 'rb').read() == modified

71 if scenario == 'metadata':

72 os.setxattr(path, 'user.chj-test', b'attribute')

73 request['baseline'] = value(dict(operation='read', path=path))['baseline']

74 staged = stage()

75 result = value(request)

76 final = os.stat(path)

77 assert result['status'] == 'saved'

78 assert final.st_uid == baseline['uid'] and final.st_gid == baseline['gid']

79 assert stat.S_IMODE(final.st_mode) == 0o6750

80 assert os.getxattr(path, 'user.chj-test') == b'attribute'

81 assert open(path, 'rb').read() == modified

82 assert open(staged, 'rb').read() == modified

83 assert final.st_mtime_ns != int(baseline['mtimeNs'])

84 assert value(dict(operation='inspect', id=ident))['status'] == 'confirmed'

85 value(dict(operation='cleanup', id=ident))

86 assert not os.path.exists(staged)

87 assert not os.listdir(home + '/ch-j-sm')

88 elif scenario == 'move-failure':

89 staged = stage()

90 with patch.object(os, 'replace', side_effect=OSError('failed rename')):

91 assert not command(request)['ok']

92 intact(staged)

93 assert os.listdir(parent) == [os.path.basename(path)]

94 elif scenario == 'disk-full':

95 staged = stage()

96 with patch.object(os, 'fsync', side_effect=OSError('disk full')):

97 assert not command(request)['ok']

98 intact(staged)

99 elif scenario == 'post-rename-disconnect':

100 staged = stage()

101 real_replace = os.replace

102

103 def replace(*args, **kwargs):

104 real_replace(*args, **kwargs)

105 # Failure to confirm after the actual rename must be reported unknown.

106 real_fsync = os.fsync

107 count = [0]

108

109 def fsync(fd):

110 count[0] += 1

111 if count[0] == 3:

112 raise OSError('lost completion')

113 return real_fsync(fd)

114 with patch.object(os, 'replace', replace), patch.object(os, 'fsync', fsync):

115 result = command(request)

116 assert result['code'] == 'FILE_RESULT_UNKNOWN', result

117 assert open(path, 'rb').read() == modified

118 assert open(staged, 'rb').read() == modified

119 assert value(dict(operation='list'))['items'][0]['status'] == 'confirmed'

120 elif scenario == 'conflict':

121 staged = stage()

122 with open(path, 'wb') as out:

123 out.write(b'external\n')

124 assert command(request)['code'] == 'FILE_CONFLICT'

125 assert open(path, 'rb').read() == b'external\n'

126 assert open(staged, 'rb').read() == modified

127 assert value(dict(operation='inspect', id=ident))['status'] == 'conflict'

128 assert command(dict(operation='cleanup', id=ident))['code'] == 'FILE_RESULT_UNCONFIRMED'

129 assert os.path.exists(staged)

130 elif scenario == 'upload-incomplete':

131 staged = stage(b'part')

132 assert command(request)['code'] == 'FILE_UPLOAD_INCOMPLETE'

133 assert open(path, 'rb').read() == original

134 assert value(dict(operation='list'))['items'][0]['status'] == 'upload-incomplete'

135 elif scenario == 'existing-temporary':

136 staged = stage()

137 assert not command(dict(operation='prepare', id=ident, path=path, baseline=baseline, hash=digest))['ok']

138 intact(staged)

139 elif scenario in ('target-hardlink', 'target-fifo'):

140 os.unlink(path)

141 other = parent + '/other'

142 with open(other, 'wb') as out:

143 out.write(original)

144 if scenario == 'target-hardlink':

145 os.link(other, path)

146 assert command(dict(operation='read', path=path))['code'] == 'FILE_HARDLINK_UNSUPPORTED'

147 else:

148 os.mkfifo(path)

149 assert not command(dict(operation='read', path=path))['ok']

150 assert open(other, 'rb').read() == original

151 elif scenario in ('target-symlink', 'symlink-retarget'):

152 # sites-enabled/x.conf -> ../sites-available/x.conf: edit the target, keep the link.

153 available = root + '/sites-available'

154 os.mkdir(available, 0o755)

155 real = available + '/example.conf'

156 with open(real, 'wb') as out:

157 out.write(original)

158 os.unlink(path)

159 os.symlink('../sites-available/example.conf', path)

160 opened = value(dict(operation='read', path=path))

161 assert opened['link'] is True and opened['resolvedPath'] == real

162 assert base64.b64decode(opened['data']) == original

163 request['baseline'] = opened['baseline']

164 request['resolvedPath'] = opened['resolvedPath']

165 staged = stage()

166 if scenario == 'symlink-retarget':

167 other = available + '/other.conf'

168 with open(other, 'wb') as out:

169 out.write(original)

170 os.unlink(path)

171 os.symlink(other, path)

172 assert command(request)['code'] == 'FILE_CONFLICT'

173 assert open(real, 'rb').read() == original

174 else:

175 result = value(request)

176 assert result['status'] == 'saved' and result['resolvedPath'] == real

177 assert os.path.islink(path) and os.readlink(path) == '../sites-available/example.conf'

178 assert open(real, 'rb').read() == modified

179 assert sorted(os.listdir(available)) == ['example.conf']

180 elif scenario == 'directory-symlink':

181 os.symlink(parent, root + '/linked-dir')

182 via = root + '/linked-dir/' + os.path.basename(path)

183 opened = value(dict(operation='read', path=via))

184 assert opened['link'] is True and opened['resolvedPath'] == path

185 request.update(path=via, baseline=opened['baseline'], resolvedPath=opened['resolvedPath'])

186 stage()

187 assert value(request)['status'] == 'saved'

188 assert open(path, 'rb').read() == modified and os.path.islink(root + '/linked-dir')

189 elif scenario == 'no-space':

190 staged = stage()

191 real_fdopen = os.fdopen

192

193 class Full:

194 def __init__(self, fd, *args, **kwargs):

195 self.inner = real_fdopen(fd, *args, **kwargs)

196 def write(self, data):

197 raise OSError(errno.ENOSPC, 'No space left on device')

198 def __getattr__(self, name):

199 return getattr(self.inner, name)

200 def __enter__(self):

201 return self

202 def __exit__(self, *args):

203 self.inner.close()

204 with patch.object(os, 'fdopen', Full):

205 assert command(request)['code'] == 'FILE_NO_SPACE'

206 intact(staged)

207 assert os.listdir(parent) == [os.path.basename(path)]

208 elif scenario == 'xattr-unsupported':

209 with patch.object(os, 'listxattr', side_effect=OSError(errno.ENOTSUP, 'Operation not supported')):

210 opened = value(dict(operation='read', path=path))

211 assert opened['baseline']['xattrs'] == {}

212 request['baseline'] = opened['baseline']

213 stage()

214 assert value(request)['status'] == 'saved'

215 assert open(path, 'rb').read() == modified

216 elif scenario == 'trusted-xattr':

217 # Root-only attributes (XFS ACL mirror, overlay metadata) are missing from a baseline captured by

218 # the SSH account; saving with sudo must neither report a conflict nor drop them.

219 if os.geteuid() != 0:

220 print('trusted.* attributes require root')

221 sys.exit(77)

222 try:

223 os.setxattr(path, 'trusted.chj-test', b'root-only')

224 except PermissionError:

225 print('trusted.* needs CAP_SYS_ADMIN (not available in this container)')

226 sys.exit(77)

227 full = value(dict(operation='read', path=path))['baseline']

228 user_view = dict(full, xattrs={k: v for k, v in full['xattrs'].items() if not k.startswith('trusted.')})

229 request['baseline'] = user_view

230 stage()

231 assert value(request)['status'] == 'saved'

232 assert os.getxattr(path, 'trusted.chj-test') == b'root-only'

233 assert open(path, 'rb').read() == modified

234 elif scenario == 'backup':

235 stage()

236 result = value(dict(request, backup=True))

237 assert open(result['backupPath'], 'rb').read() == original

238 assert stat.S_IMODE(os.stat(result['backupPath']).st_mode) == 0o600

239 assert open(path, 'rb').read() == modified

240 elif scenario.startswith('publish-') or scenario == 'stage-out':

241 payload = os.urandom(3 * 1024 * 1024 + 17)

242 digest = hashlib.sha256(payload).hexdigest()

243 os.setxattr(path, 'user.chj-test', b'kept')

244 os.chmod(path, 0o640)

245 uid = os.getuid()

246 target_stat = value(dict(operation='stat', path=path))

247 assert target_stat['exists'] and target_stat['type'] == 'file'

248 direct = parent + '/.part-direct'

249 fd = os.open(direct, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

250 with os.fdopen(fd, 'wb') as out:

251 out.write(payload)

252 publish = dict(operation='publish', source=direct, uid=uid, path=path, hash=digest, size=len(payload),

253 baseline=target_stat['baseline'], strategy='rename')

254 if scenario == 'publish-existing-rename':

255 result = value(publish)

256 assert result['status'] == 'published' and result['metadataPreserved'] and not result['created']

257 assert open(path, 'rb').read() == payload

258 assert os.getxattr(path, 'user.chj-test') == b'kept' and stat.S_IMODE(os.stat(path).st_mode) == 0o640

259 assert not os.path.exists(direct)

260 elif scenario == 'publish-new-rename':

261 fresh = parent + '/fresh.conf'

262 result = value(dict(publish, path=fresh, baseline=None, mode=0o640))

263 assert result['created'] and open(fresh, 'rb').read() == payload

264 assert stat.S_IMODE(os.stat(fresh).st_mode) == 0o640 and not os.path.exists(direct)

265 elif scenario == 'publish-copy':

266 ws = value(dict(operation='transfer-workspace'))['path']

267 staged = ws + '/' + 'b' * 32 + '.part'

268 fd = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

269 with os.fdopen(fd, 'wb') as out:

270 out.write(payload)

271 nested = root + '/new parent/deeper/file.bin'

272 created = value(dict(publish, source=staged, path=nested, baseline=None, strategy='copy', parents=True))

273 assert created['created'] and created['sourceRemoved'] and open(nested, 'rb').read() == payload

274 assert not os.path.exists(staged)

275 assert [n for n in os.listdir(os.path.dirname(nested))] == ['file.bin']

276 fd = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

277 with os.fdopen(fd, 'wb') as out:

278 out.write(payload)

279 replaced = value(dict(publish, source=staged, strategy='copy'))

280 assert replaced['metadataPreserved'] and open(path, 'rb').read() == payload

281 assert os.getxattr(path, 'user.chj-test') == b'kept'

282 elif scenario == 'publish-conflict':

283 with open(path, 'ab') as out:

284 out.write(b'external')

285 assert command(publish)['code'] == 'FILE_CONFLICT'

286 assert open(path, 'rb').read() == original + b'external' and os.path.exists(direct)

287 elif scenario == 'publish-verify':

288 assert command(dict(publish, hash='0' * 64))['code'] == 'FILE_VERIFY_FAILED'

289 assert open(path, 'rb').read() == original

290 elif scenario == 'publish-exists':

291 assert command(dict(publish, baseline=None))['code'] == 'FILE_EXISTS'

292 assert value(dict(publish, baseline=None, overwrite=True))['status'] == 'published'

293 elif scenario == 'publish-retry':

294 value(publish)

295 again = value(dict(publish, retry=True))

296 assert again['alreadyPublished'] is True

297 elif scenario == 'stage-out':

298 with open(path, 'wb') as out:

299 out.write(payload)

300 staged = value(dict(operation='stage-out', path=path, id='c' * 32, uid=uid))

301 assert staged['hash'] == digest and staged['size'] == len(payload)

302 assert open(staged['path'], 'rb').read() == payload

303 assert stat.S_IMODE(os.stat(staged['path']).st_mode) == 0o600

304 listed = value(dict(operation='transfer-list'))['items']

305 assert [item['id'] for item in listed] == ['c' * 32]

306 assert value(dict(operation='hash', path=staged['path']))['hash'] == digest

307 assert value(dict(operation='transfer-remove', id='c' * 32))['removed'] is True

308 assert not os.path.exists(staged['path'])

309 else:

310 raise AssertionError('Unknown scenario: ' + scenario)

311 elif scenario == 'file-ops':

312 work = root + '/ops'

313 value(dict(operation='mkdir', path=work + '/a/b', parents=True))

314 value(dict(operation='create', path=work + '/a/b/new.txt', mode=0o600))

315 assert stat.S_IMODE(os.stat(work + '/a/b/new.txt').st_mode) == 0o600

316 assert command(dict(operation='create', path=work + '/a/b/new.txt'))['code'] == 'FILE_EXISTS'

317 os.symlink('b', work + '/a/link')

318 listed = value(dict(operation='listdir', path=work + '/a'))['entries']

319

        assert {e['name']: (e['type'], e.get('targetType')) for e in listed} == {'b': ('directory', None), 'link': ('symlink', 'directory')}

320 value(dict(operation='create', path=work + '/a/other.txt'))

321

        assert command(dict(operation='rename', **{'from': work + '/a/b/new.txt', 'to': work + '/a/other.txt'}))['code'] == 'FILE_EXISTS'

322 value(dict(operation='rename', **{'from': work + '/a/b/new.txt', 'to': work + '/a/renamed.txt'}))

323 assert os.path.exists(work + '/a/renamed.txt')

324 value(dict(operation='remove', path=work + '/a/link'))

325 assert os.path.isdir(work + '/a/b') and not os.path.lexists(work + '/a/link')

326 assert command(dict(operation='remove', path=work + '/a'))['code'] == 'FILE_DIRECTORY_NOT_EMPTY'

327 removed = value(dict(operation='remove', path=work + '/a', recursive=True))

328 assert removed['entries'] >= 3 and not os.path.exists(work + '/a')

329 elif scenario == 'workspace-symlink':

330 os.symlink(parent, home + '/ch-j-sm')

331 assert not command(dict(operation='prepare', id=ident, path=path, baseline=baseline, hash=digest))['ok']

332 assert os.listdir(parent) == [os.path.basename(path)]

333 elif scenario == 'temporary-symlink':

334 staged = stage()

335 os.unlink(staged)

336 os.symlink(path, staged)

337 assert not command(request)['ok']

338 assert open(path, 'rb').read() == original

339 elif scenario == 'metadata-failure':

340 os.setxattr(path, 'user.chj-test', b'attribute')

341 request['baseline'] = value(dict(operation='read', path=path))['baseline']

342 staged = stage()

343 with patch.object(os, 'setxattr', side_effect=PermissionError('metadata denied')):

344 assert not command(request)['ok']

345 intact(staged)

346 elif scenario == 'acl':

347 # Linux POSIX ACL xattr format: version, followed by tag/perm/uid entries.

348 acl = struct.pack('<I', 2) + b''.join(struct.pack('<HHI', tag, perm, who) for tag, perm, who in [

349 (1, 7, 0xffffffff), (2, 4, os.getuid() + 1), (4, 5, 0xffffffff), (16, 5, 0xffffffff), (32, 0, 0xffffffff)])

350 try:

351 os.setxattr(path, 'system.posix_acl_access', acl)

352 except OSError:

353 print('POSIX ACLs are unavailable on the test filesystem')

354 sys.exit(77)

355 request['baseline'] = value(dict(operation='read', path=path))['baseline']

356 staged = stage()

357 assert value(request)['status'] == 'saved'

358 assert os.getxattr(path, 'system.posix_acl_access') == acl

359 assert open(staged, 'rb').read() == modified

360 elif scenario == 'concurrent-target-replacement':

361 staged = stage()

362 real_fsync = os.fsync

363 count = [0]

364

365 def fsync(fd):

366 count[0] += 1

367 if count[0] == 2:

368 replacement = parent + '/external'

369 with open(replacement, 'wb') as out:

370 out.write(b'external')

371 os.replace(replacement, path)

372 return real_fsync(fd)

373 with patch.object(os, 'fsync', fsync):

374 assert command(request)['code'] == 'FILE_CONFLICT'

375 assert open(path, 'rb').read() == b'external'

376 assert open(staged, 'rb').read() == modified

377 elif scenario == 'cross-filesystem-strategy':

378 staged = stage()

379 assert os.stat(staged).st_dev != os.stat(path).st_dev

380 real_replace = os.replace

381

382 def replace(src, dst, **kwargs):

383 assert src != staged and '/' not in src

384 assert kwargs['src_dir_fd'] == kwargs['dst_dir_fd']

385 return real_replace(src, dst, **kwargs)

386 with patch.object(os, 'replace', replace):

387 assert value(request)['status'] == 'saved'

388 assert os.path.exists(staged)

389 else:

390 raise AssertionError('Unknown scenario: ' + scenario)

391

392print('ok')

393

SHA-256

c6684dbafe2f8a6bc9aa39a3151bd13e52836aded427b1157c92855e11cf635a

CH-J Proprietary Software License 1.14 · Kode sumber disediakan berdasarkan CH-J Proprietary Software License 1.14. Ketersediaannya tidak mengubah ketentuan lisensi atau memberikan hak tambahan.

SHA-256 arsip

a9605c0870a223b6a8542bb0705edd51d8b04f3069582653417a8397fabcc09a