공개된 소스

CH-J Server Manager

특정 애플리케이션 릴리스의 디렉터리와 파일을 살펴보세요.

소스 ZIP 다운로드

CH-J / Server Manager 193 개 파일

core-20261011T184039Z-4148710795d0 추가됨: 날짜 기록 없음 · 빌드됨: 2026-10-11 20:40

remoteEditorLinux.py 추가됨: 날짜 기록 없음 9.2 KB · 214 줄

코드 파일 다운로드

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 hashlib

9import io

10import json

11import os

12import pwd

13import stat

14import struct

15import sys

16import tempfile

17from types import SimpleNamespace

18from unittest.mock import patch

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

21scenario = sys.argv[2]

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

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

25 home = root + '/actual home'

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

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

28 print('No separate writable filesystem available')

29 sys.exit(77)

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

31 os.mkdir(home, 0o700)

32 parent = root + '/target directory'

33 os.mkdir(parent, 0o700)

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

35 original = b'original\n'

36 modified = b'modified\n'

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

38 out.write(original)

39 os.chmod(path, 0o6750)

41 def command(request):

42 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):

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

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

47 def value(request):

48 result = command(request)

49 assert result['ok'], result

50 return result['value']

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

53 ident = 'a' * 32

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

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

57 def stage(data=modified):

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

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

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

61 out.write(data)

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

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

64 return prepared['temporary']

66 def intact(staged):

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

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

70 if scenario == 'metadata':

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

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

73 staged = stage()

74 result = value(request)

75 final = os.stat(path)

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

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

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

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

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

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

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

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

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

85 assert not os.path.exists(staged)

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

87 elif scenario == 'move-failure':

88 staged = stage()

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

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

91 intact(staged)

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

93 elif scenario == 'disk-full':

94 staged = stage()

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

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

97 intact(staged)

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

99 staged = stage()

100 real_replace = os.replace

101

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

103 real_replace(*args, **kwargs)

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

105 real_fsync = os.fsync

106 count = [0]

107

108 def fsync(fd):

109 count[0] += 1

110 if count[0] == 3:

111 raise OSError('lost completion')

112 return real_fsync(fd)

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

114 result = command(request)

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

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

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

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

119 elif scenario == 'conflict':

120 staged = stage()

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

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

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

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

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

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

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

128 assert os.path.exists(staged)

129 elif scenario == 'upload-incomplete':

130 staged = stage(b'part')

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

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

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

134 elif scenario == 'existing-temporary':

135 staged = stage()

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

137 intact(staged)

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

139 os.unlink(path)

140 other = parent + '/other'

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

142 out.write(original)

143 if scenario == 'target-symlink':

144 os.symlink(other, path)

145 elif scenario == 'target-hardlink':

146 os.link(other, path)

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 == 'workspace-symlink':

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

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

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

155 elif scenario == 'temporary-symlink':

156 staged = stage()

157 os.unlink(staged)

158 os.symlink(path, staged)

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

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

161 elif scenario == 'metadata-failure':

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

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

164 staged = stage()

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

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

167 intact(staged)

168 elif scenario == 'acl':

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

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

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

172 try:

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

174 except OSError:

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

176 sys.exit(77)

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

178 staged = stage()

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

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

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

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

183 staged = stage()

184 real_fsync = os.fsync

185 count = [0]

186

187 def fsync(fd):

188 count[0] += 1

189 if count[0] == 2:

190 replacement = parent + '/external'

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

192 out.write(b'external')

193 os.replace(replacement, path)

194 return real_fsync(fd)

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

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

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

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

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

200 staged = stage()

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

202 real_replace = os.replace

203

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

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

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

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

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

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

210 assert os.path.exists(staged)

211 else:

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

213

214print('ok')

215

SHA-256

a42aa6364631ccc6bc0cb1ff5b8aee7ee51620d56c0a41f47b3cd981c33f5d99

CH-J Proprietary Software License 1.14 · 소스는 CH-J Proprietary Software License 1.14에 따라 제공됩니다. 소스를 공개한다고 해서 라이선스 조항이 바뀌거나 추가 권리가 부여되지는 않습니다.

압축 파일 SHA-256

5a8ea02d06a5643cf2330fe9f54b8767a06c1c59d36527705e536fa9dbca7614