信頼はずっと、挑戦はもっと。

お問い合わせ
TEL:03-3496-3888

BLOG コアテックの社員ブログ (毎週月曜~金曜更新中)

LIST OF ARTICLES

記事一覧

  • 画像:ブログサムネイル

    テクログ

    【nfcpyでLINE通知】ちょうど目の前にICカードリーダーとNFCチップが転がっていたので

    こんにちはじゅんすです。私は実家暮らしで兄弟がいるのですが、どうも兄が家に帰ってきても手を洗っていないっぽいのです。時期が時期なのでちゃんと手洗いうがいはしてほしいのですが、、。こりゃどうしたもんか、、、と思ったその時、ちょうど目の前にICカードリーダーとNFCチップが転がっていたので「手を洗ったらICカードリーダーにピッてしてLINEのトークルームに通知を送るシステム」を作っちゃいました。Pythonだと参考資料が多かったのでPythonで開発しました。触ったことないですけどね、、資料が多かったもんで、。なのでコードに至らない部分は多くあると思いますがご了承ください。以下にざっくりと処理の流れを記載しておきます。ICカードリーダーでNFCチップのIDm(固有ID)を取得。取得したIDmでDBからユーザー名を取得。「〇〇〇(ユーザー名)は手を洗ったよ」という内容でLINE通知。開発環境■各バージョンWindows 10 Home 64-bitPython 3.8.2libusb 1.0.23Zadig 2.5nfcpy 1.0.3MySQL 8.0.19■DBクライアントMySQL Workbench■USBドライバZadig■ICカードリーダー非接触型ICカードリーダー / ライター RC-S380(SONY製)■NFCチップNTAG213(NXP製)下準備ただICカードリーダー差してコードを書いただけじゃ動かないので以下の準備をしておきます。■libusbのインストール公式サイト(https://libusb.info/)からダウンロードできます(私の時はlibusb-1.0.23.7zでした)。ダウンロードしたファイルを解凍したら以下の操作をします。64ビット版Windowsの場合MS64\dll\libusb-1.0.dllをC:\Windows\System32にコピーMS32\dll\libusb-1.0.dllをC:\Windows\SysWOW64にコピー32ビット版Windowsの場合MS32\dll\libusb-1.0.dllをC:\Windows\System32にコピー■ドライバの適用公式サイト(https://zadig.akeo.ie/)からダウンロードできます(私の時はZadig 2.5でした)。ダウンロードしたらZadigアプリケーションを実行する前にPCにRC-S380を差します。ドライバが自動適用されたらZadigアプリケーションを実行します。表示されたウィンドウの上にあるリストボックスからNFC Port/PaSoRi 100 USBを選択します(ない場合はOptions ⇒ List All Devicesで表示されます)。Driverの項目はWinUSBにします。Replace Driverをクリックしたら完了です。■LINE Notify アクセストークンの発行公式サイト(https://notify-bot.line.me/ja/)にログインして、右上のアカウントからマイページに行きます。マイページの下部にトークンを発行するがあるのでトークン名の入力と通知したいトークルームを選択して発行します。※発行されたトークンは1度しか表示されないので注意。■DBの設定まずhand_washingという名前でDBを作成します。CREATE DATABASE hand_washing  CHARACTER SET utf8  COLLATE utf8_general_ci; 続いてNFCチップのデータを入れるnfc_tipsと、それを使うユーザーのデータを入れるusersテーブルを作成します。CREATE TABLE `hand_washing`.`nfc_tips` (  `id` INT NOT NULL AUTO_INCREMENT,  `idm` VARCHAR(50) NOT NULL COMMENT 'NFCチップの固有ID',  `user_id` INT NOT NULL COMMENT 'NFCチップを使用するユーザーのID',  `created_at` DATETIME NOT NULL,  `updated_at` DATETIME NOT NULL,  PRIMARY KEY (`id`)) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8 COMMENT = 'NFCチップデータ用のテーブル'; CREATE TABLE `hand_washing`.`users` (  `id` INT NOT NULL AUTO_INCREMENT,  `name` VARCHAR(10) NOT NULL COMMENT 'ユーザー名',  `created_at` DATETIME NOT NULL,  `updated_at` DATETIME NOT NULL,  PRIMARY KEY (`id`)) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8 COMMENT = 'NFCチップを使用するユーザー用テーブル'; テーブル作成も無事終わったらそれぞれのテーブルにデータを登録しておきます。nfc_tipsのidmには後述するnfc_reader.pyで取得したidmを登録します(ちょっと手間ですが、、登録機能を先に作っといた方がよかったなぁ)。動作確認■パッケージのインストールnfcpyとrequestsをpipでインストールします。pip install nfcpy pip install requests ■ファイルの作成以下3つのファイルを全て同階層に作成します。DBに接続してNFCチップのIDmからユーザー名を取得する処理を書いたdb_crud.py。import MySQLdb # データベース接続 connection = MySQLdb.connect(  host = 'ホスト名もしくはIP',  user = 'ユーザー名',  passwd = 'パスワード',  db = 'hand_washing',  charset='utf8' ) # カーソル生成 cursor = connection.cursor() # NFCのIDmからユーザー名を取得 def getUserNameByIdm(idm):  cursor.execute("SELECT users.name FROM nfc_tips JOIN users ON nfc_tips.user_id = users.id WHERE idm = '%s'" % idm.decode())  return cursor.fetchone()[0] LINEに通知を送る処理を書いたline_notify.py。import requests # LINE Notifyの情報 line_notify_token = 'アクセストークン' line_notify_api = 'https://notify-api.line.me/api/notify' # LINEでメッセージを送る def sendMessage(person):  message = str(person) + 'は手を洗ったよ'  payload = {'message': message}  headers = {'Authorization': 'Bearer ' + line_notify_token}  # 送信  requests.post(line_notify_api, data = payload, headers = headers) NFCチップのデータを読み込む処理を書いたnfc_reader.py。import nfc import binascii import time # 外部ファイル import db_crud import line_notify # タッチ待ち受けの1サイクル秒 TIME_CYCLE = 2.0 # タッチ待ち受けの反応インターバル秒 TIME_INTERVAL = 2.0 # タッチされてから次の待ち受けを開始するまで無効化する秒 TIME_WAIT = 5 # NFC接続リクエストのための準備 # NXP製のNFCチップのため106A(通信規格TypeA)で設定(Suicaで試したい人は212F) remoteTarget = nfc.clf.RemoteTarget("106A") print("カードをかざしてください") while True:  # USBに接続されたNFCリーダに接続してインスタンス化  with nfc.ContactlessFrontend("usb") as clf:   # タッチ待ち受け開始   # clf.sense( [リモートターゲット], [検索回数], [検索の間隔] )   target = clf.sense(remoteTarget, iterations = int(TIME_CYCLE//TIME_INTERVAL) + 1, interval = TIME_INTERVAL)   while target:    tag = nfc.tag.activate(clf, target)    if tag != None:     #固有のIDmを取り出す     idm = binascii.hexlify(tag.identifier).upper()     # ユーザー名取得     person = db_crud.getUserNameByIdm(idm)     print("読み込み成功\n")     print("タグ情報:" + str(tag))     print("IDm:" + idm.decode())     print("ユーザー名:" + person + "\n")     print("カードをかざしてください")     # LINEに通知する     line_notify.sendMessage(person)    else:     print("読み込みに失敗しました")    time.sleep(TIME_WAIT)    break 各ファイルが用意出来たらターミナルでnfc_reader.pyのある階層までいきます。以下のコマンドでnfc_reader.pyの処理を実行します。python nfc_reader.py すると以下のように表示されるのでNFCチップをICカードリーダーに2秒ほどかざします。python nfc_reader.py カードをかざしてください NFCチップのデータとユーザー名が表示されると同時にLINEに通知が送られます。python nfc_reader.py カードをかざしてください タグ情報:〇〇〇〇〇〇 IDm:〇〇〇〇〇〇 ユーザー名:〇〇〇〇〇〇 カードをかざしてください とまぁ、こんな感じで通知システムを作ってみたってお話です。ただ、これだけじゃ手を洗ったふりしてICカードリーダーにピッてする不正行為ができてしまうので、手の汚れをブラックライトで可視化してカメラで読み込む機能等と組み合わせるなどしたほうが良いかなとか思ったり思わなかったりって感じですね、、(Raspberry Piが欲しい)。皆さんも目の前に転がっていたら色々作ってみてください!ではでは!
  • テクログ

    【E2Eテスト】codeceptjsをテスト環境に導入してから数ヶ月経過

    こんにちは。E2Eテストを導入して数ヶ月経過しました。画面が生きているかの確認や一定のアクションを実行できるかの確認程度のテストを常に実施しています。忘れた頃にテスト失敗通知がChatWorkに届きます。実施していなかった頃に比べると「検知」のスピードは格段に上がっていると感じます。このままテストと仲良くなりたいです。
  • 画像:ブログサムネイル

    テクログ

    wafのサンプルログを定期的にとって、あったら自動でchatworkに通知してほしいというきもち

    どうも、相変わらずいろいろ検証をしています。RDS Proxy はやくGAになってほしいですね。さて、標題の気持ちになったのでLambdaを書きました。ファイル分け?大丈夫大丈夫このくらいなら。冗長だとしてもわかりやすさ優先なのはいつもどおりです。弊社ではcloud9でのserverless frameworkでのデプロイも普通になってきましたね。Lambdaがはかどりますね。さてソースです。こちら、参考URLです。https://dev.classmethod.jp/articles/get-aws-waf-sample-logs/こちらを色々変更し、指定したwaf 指定したルールを 指定した間隔で行うようにしています。環境変数に入るだけ何個でもいけますね。あ、これはCF版ですが、LB版は上のクライアントを変えるだけだと思います。(作って運用しています。増えすぎるのであえてCF版とLB版を分けました)実行はcloudwatch eventsで設定です。slsの設定で。環境変数はこんな感じCHATWORK_API_TOKEN XXあんごうかしたトークンだよXXCHECK_LISTS [{'Name':'ばいたいめいだよ','WebAclID':'aaaa-aaaaとかだよ','WAFName':'わかりやすくするためのwafの名前です','RuleID':'aaaa-bbbbとかのルールID','RuleName':'これも基本わかりやすくするためのルール名'},]INTERVAL 11(とか数字)ROOM_ID 11111111(とかのルームID)# -*- coding: utf-8 -*- #cfのwafのサンプルログを確認し、あれば通知する import os import urllib from datetime import datetime, timedelta import boto3 import sys from base64 import b64decode import ast REGION = "ap-northeast-1" #環境変数 #Chatwork API トークン ENCRYPTED_CHATWORK_API_TOKEN = os.environ['CHATWORK_API_TOKEN'] CHATWORK_API_TOKEN = boto3.client('kms').decrypt(CiphertextBlob=b64decode(ENCRYPTED_CHATWORK_API_TOKEN))['Plaintext'] INTERVAL = int(os.environ['INTERVAL']) ROOM_ID = os.environ['ROOM_ID'] CHECK_LISTS = ast.literal_eval(os.environ['CHECK_LISTS']) #client用意 CF版だよ waf = boto3.client('waf') #### # replace datetime to string #### def time2str(x):     x['Timestamp'] = x['Timestamp'].isoformat()     return x #### #chatworkに通知 #### def postChatwork(message, room_id):   END_POINT_BASE = "https://api.chatwork.com/v2"   END_POINT      = "/rooms/"   ACTION         = "/messages"      headers = { 'X-ChatWorkToken': CHATWORK_API_TOKEN }      data  = { 'body': message }   data = urllib.parse.urlencode(data)   data = data.encode('utf-8')      # リクエストの生成と送信   post_message_url = END_POINT_BASE + END_POINT + room_id + ACTION   request = urllib.request.Request(post_message_url, data=data, method="POST", headers=headers)   with urllib.request.urlopen(request) as response:     response_body = response.read().decode("utf-8")     print(response_body) #### #dictを文字にするだけ #### def makeText(marge_logs):   text = "[info][title]CF版 WAF 検知ログ 直近"+str(INTERVAL)+"分[/title]"      for log in marge_logs:     for key, value in log.items():       text += key + "    " + value + "\n"     text += "[hr]"        text += "[/info]"        return text ### #main ### def lambda_handler(event, context):   #LISTごとに繰り返し   for CHECK_LIST in CHECK_LISTS:     WebAclId = CHECK_LIST['WebAclID']     acl = waf.get_web_acl(WebACLId=WebAclId)          WebACLName = acl['WebACL']['Name']        marge_logs = []     for rule in acl['WebACL']['Rules']:       RuleId = rule['RuleId']              if RuleId == CHECK_LIST['RuleID']:         # get sample requests         r = waf.get_sampled_requests(           MaxItems=3,           WebAclId=WebAclId,           RuleId=RuleId,           TimeWindow={             'EndTime': datetime.utcnow(),             'StartTime': datetime.utcnow() - timedelta(minutes=INTERVAL)           }         )              # check sample count         if 'SampledRequests' not in r or len(r['SampledRequests']) == 0:           #ログのみ           print("WebAclId:" + WebAclId + " RuleId:" +RuleId + " サンプルカウントなし")         else:           # replace datetime to string           logs = list(map(time2str, r['SampledRequests']))           for l in logs:             headers_dict = l['Request']['Headers']                      Host = ''             From = ''             UserAgent = ''             UserAgent2 = ''             Referer = ''             Referer2 = ''             for hd in headers_dict:               if hd['Name'] == 'Host':                 Host = hd['Value']               elif hd['Name'] == 'From':                 From = hd['Value']               elif hd['Name'] == 'User-Agent':                 UserAgent = hd['Value']               elif hd['Name'] == 'user-agent':                 UserAgent2 = hd['Value']               elif hd['Name'] == 'Referer':                 Referer = hd['Value']               elif hd['Name'] == 'referer':                 Referer2 = hd['Value']                          simpledict = {               '媒体名':CHECK_LIST['Name'],               '時間':l['Timestamp'],               'IP':'https://ipee.at/'+l['Request']['ClientIP'],               '国':l['Request']['Country'],               'URI':l['Request']['URI'],               'Headers.Host':Host,               'Headers.From':From,               'Headers.User-Agent':UserAgent + UserAgent2,               'Headers.Referer':Referer + Referer2,               '対象WAF/ルール':CHECK_LIST['WAFName']+' / '+CHECK_LIST['RuleName'],               'Method':l['Request']['Method'],             }             marge_logs.append(simpledict)               # chatwork通知           if marge_logs != []:             try:               print(marge_logs)               cw_text = makeText(marge_logs)               postChatwork(cw_text, ROOM_ID)             except Exception as e:               print(e)           else:             postChatwork("ログないよ", ROOM_ID)#基本ここはないですね ではよいWAF生活を!
  • テクログ

    ルーター異常? 調査記録

    現在グループ会社でDHCPが以上に使われる現象が起きています。ルーターでDHCPの枠は130ありますが1日経つと空きが0になっています。wifiのせいかと思い別回線に移動させており今まで圧迫していたIPを開放しました。現在社内のIPは固定化しておりDHCPの設定外にあるので足り無くなるはずはないのですが。またVPNソフトを使い外部からIPを取ってはいますが、20人もいないはずなので0にはならないはず。まとめ・社内は固定IPに設定・wifiはDHCPだが別回線で繋いでいるのでメイン回線には関係がない・VPNでのDHCPは20人程度・社内でテストなどDHCPを使う事があっても100は埋まらない上記を踏まえて今後の点検作業としては・ルーターのDHCP開放がうまくいっていない・管理者が把握していないDHCPを発行する機器がある(個人wifiなど)ますは以上2点に絞り機器の点検を行う予定です。
  • テクログ

    【AWS】ELBのログをS3に保存して、Athenaで検索しよう!

    こんにちは皆さんAWS使ってますか?LB、使ってます??AWSのLBもいくつか種類がありますねhttps://docs.aws.amazon.com/elasticloadbalancing/index.html弊社も大変お世話になっていますそんなこんなでLBのアクセスログをS3に保存するととてもいいことがあります。https://tech.basicinc.jp/articles/46詳しいやり方はこちらのサイト様を参考にして頂いて・・・S3に保存するとAthenaでSQLを使って検索が出来るようになるので非常に便利です。よく使うクエリはこんな感じです。○該当IPがアクセスしているURLを検索するselect  request_ip, url from  ELBログの保存先 where  elb_name = ‘たくさんあると絞ってあげたほうがわかりやすいです’   AND  (regexp_like(request_ip, ‘111.111.111.111’)) limit 1000; こんな感じで○該当URLにアクセスしているIPを割り出す○該当URLにアクセスしているIPのカウント数○該当日時にアクセスしてきたIPやURLを割り出すなどが検索可能です。更に応用としてはUAでGoogleBotがクローリングしているURLを割り出したり(もちろんUAは偽装できてしまうので正確とは言えないですが・・・参考までに)日毎のGoogleBotの件数を見てみたり(↓)SELECT substr(request_timestamp,      1,     10) AS ymd,      count(*) AS cnt FROM ELBログの保存先 WHERE elb_name = ‘たくさんあると絞ってあげたほうがわかりやすいです’     AND user_agent LIKE '%Googlebot%' GROUP BY substr(request_timestamp, 1,10) ORDER BY ymd; あとはprocessing_time(何種類かあります)でorder by descすれば処理が重いページもわかったりします。便利ですね!!皆さんもELBのログをS3に保存して素晴らしいAthenaライフをお送りください!!
  • テクログ

    LINE Notify APIを試してみました

    初めまして。1月からお世話になっておりますJGです。最近LINE Notify APIを使い、毎朝自分のLINEに通知させました。それについて触れたいと思います。下記URLからLineIDでログインし、必要事項を入力します。その後、アクセストークン発行をクリックし、発行されたトークンをコピーして保存してください。https://notify-bot.line.me/ja/天気予報を取得する処理は下記のlivedoorを使用しました。http://weather.livedoor.com/weather_hacks/webservicePHPで処理を書き、作成したものをHerokuに上げ、毎朝通知するようにHerokuのスケジューラを設定しました。下記ソースのTOKENに、LINEのアクセストークンを指定すれば動きます。自分の住んでいる東京と出身地の神奈川の天気を通知するために、$city_listを下記のように指定しています。興味がある方は、是非試してみてください。<?php define('TOKEN', ''); define('MESSAGE_NOTIFY_API_URL', 'https://notify-api.line.me/api/notify'); define('WEATHER_API_URL', 'http://weather.livedoor.com/forecast/webservice/json/v1?city='); $city_list = [   130010,   140010,   140020 ]; foreach ($city_list as $city) {   //天気予報を取得   $weather_data = get_weather_data($city);   if (empty($weather_data)) {     $msg = "天気予報取得失敗です。";   } else {     //通知メッセージを作成     $msg = create_message($weather_data);   }   //通知   send_msg($msg); }   function get_weather_data($city) {   $ch = curl_init();   curl_setopt($ch, CURLOPT_URL, WEATHER_API_URL.$city);   curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'GET');   curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);   $response = curl_exec($ch);   $http_code = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);   if ($http_code != 200) {     return false;   }   if (curl_error($ch)) {     return false;   }   $weather_data = json_decode($response, true);   curl_close($ch);   return $weather_data; } function create_message($arr) {   $msg = '';   $title = $arr['title'] ?? '';   $msg .= $title."\n\n";   for ($i = 0; $i < 2; $i++) {     $date = $arr['forecasts'][$i]['date'] ?? '';     $date_label = $arr['forecasts'][$i]['dateLabel'] ?? '';     $weather = $arr['forecasts'][$i]['image']['title'] ?? '';     $min = $arr['forecasts'][$i]['temperature']['min']['celsius'] ?? '';     $max = $arr['forecasts'][$i]['temperature']['max']['celsius'] ?? '';     $msg .= $date_label.'('.$date.')'."\n";     $msg .= '天気: '.$weather."\n";     $msg .= '最高気温: '.$max."\n";     $msg .= '最低気温: '.$min;     if ($i < 1) {       $msg .= "\n\n";     }   }   return $msg; } function send_msg($msg) {   $post_data = http_build_query(['message' => $msg]);   $ch = curl_init(MESSAGE_NOTIFY_API_URL);   curl_setopt($ch, CURLOPT_POST, true);   curl_setopt($ch, CURLOPT_POSTFIELDS, $post_data);   curl_setopt($ch, CURLOPT_HTTPHEADER, [     'Content-Type: application/x-www-form-urlencoded',     'Authorization: Bearer '.TOKEN,     'Content-Length: '.strlen($post_data)   ]);   curl_exec($ch);   $http_code = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);   if ($http_code != 200) {     return false;   }   if (curl_error($ch)) {     return false;   }   curl_close($ch);   return true; }